LINUXDO_帖子随机数据库

r_by(username=username).first() if not existing: # 创建新频道记录 channel = Channel(username=username) session.add(channel) print(f"添加频道: {username}") # 提交更改 session.commit() if __name__ == "__main__": print("正在创建表...") create_tables() print("正在初始化频道...") init_channels() print("初始化完成！") 环境变量.env： # Telegram API 配置 TELEGRAM_API_ID= TELEGRAM_API_HASH= # 数据库配置 DATABASE_URL=postgresql://tg_user:你的密码@localhost:5432/tg_monitor # 默认频道配置（用逗号分隔多个频道，不需要包含 @ 符号） DEFAULT_CHANNELS=BaiduCloudDisk,tianyirigeng,Aliyun_4K_Movies # 日志级别 LOG_LEVEL=INFO model.py: from sqlalchemy import Column, Integer, String, DateTime, JSON, ARRAY, create_engine from sqlalchemy.ext.declarative import declarative_base from datetime import datetime from config import settings Base = declarative_base() class Message(Base): __tablename__ = "messages" id = Column(Integer, primary_key=True, index=True) timestamp = Column(DateTime, nullable=False) title = Column(String) description = Column(String) links = Column(JSON) # 存储各种网盘链接 tags = Column(ARRAY(String)) # 标签数组 source = Column(String) # 来源 channel = Column(String) # 频道 group_name = Column(String) # 群组 bot = Column(String) # 机器人 created_at = Column(DateTime, default=datetime.utcnow) class Credential(Base): __tablename__ = "credentials" id = Column(Integer, primary_key=True, index=True) api_id = Column(String, nullable=False) api_hash = Column(String, nullable=False) class Channel(Base): __tablename__ = "channels" id = Column(Integer, primary_key=True, index=True) username = Column(String, nullable=False) # 数据库连接配置 DATABASE_URL = settings.DATABASE_URL # 创建数据库引擎 engine = create_engine(DATABASE_URL) # 创建所有表 def create_tables(): Base.metadata.create_all(bind=engine) config.py: # 配置文件 # 这里只写骨架，后续再补充具体实现 from pydantic_settings import BaseSettings from dotenv import load_dotenv load_dotenv() # 加载 .env 文件 class Settings(BaseSettings): # Telegram API 配置 TELEGRAM_API_ID: int TELEGRAM_API_HASH: str # 数据库配置 DATABASE_URL: str POSTGRES_USER: str POSTGRES_PASSWORD: str POSTGRES_DB: str # 默认频道配置 DEFAULT_CHANNELS: str # 日志级别 LOG_LEVEL: str = "INFO" # Docker 环境标识 DOCKER_ENV: str = "false" class Config: env_file = ".env" # 指定 .env 文件 env_file_encoding = "utf-8" settings = Settings() 后台： import streamlit as st from sqlalchemy.orm import Session from models import Credential, Channel, engine st.set_page_config(page_title="后台管理", page_icon="🔧", layout="wide") st.title("后台管理") # API凭据管理 st.header("API凭据管理") with Session(engine) as session: creds = session.query(Credential).all() for cred in creds: col1, col2, col3 = st.columns([3, 5, 2]) col1.write(f"api_id: {cred.api_id}") col2.write(f"api_hash: {cred.api_hash}") if col3.button(f"删除", key=f"del_cred_{cred.id}"): session.delete(cred) session.commit() st.rerun() st.markdown("---") with st.form("add_cred_form"): api_id = st.text_input("新API ID") api_hash = st.text_input("新API Hash") submitted = st.form_submit_button("添加API凭据") if submitted and api_id and api_hash: session.add(Credential(api_id=api_id, api_hash=api_hash)) session.commit() st.success("添加成功！") st.rerun() # 频道管理 st.header("监听频道管理") with Session(engine) as session: chans = session.query(Channel).all() for chan in chans: col1, col2 = st.columns([6, 2]) col1.write(f"频道: {chan.username}") if col2.button(f"删除", key=f"del_chan_{chan.id}"): session.delete(chan) session.commit() st.rerun() st.markdown("---") with st.form("add_chan_form"): username = st.text_input("新频道用户名（不加@）") submitted = st.form_submit_button("添加频道") if submitted and username: session.add(Channel(username=username)) session.commit() st.success("添加成功！") st.rerun()
时间: Mon, 02 Jun 2025 04:26:26 +0000
作者: drfy
回复: 好想法，能提供下rss不，我刚好写了个rss聚合通知
时间: Mon, 02 Jun 2025 04:29:52 +0000
作者: dekviw
回复: 感觉有点难，搞成这样感觉已经是我的极限了，不会编程的。全靠cursor指挥，部署步骤都是磨了半天才磨出来，想搞成docker都完全弄不除出来。。。。
时间: Mon, 02 Jun 2025 04:33:28 +0000
作者: 大帅哥
回复: 太强了！
时间: Mon, 02 Jun 2025 04:42:20 +0000
作者: Zane你发财
回复: 感谢佬友的分享！ ~又抓到你了（大帅哥）
时间: Mon, 02 Jun 2025 07:36:25 +0000
作者: 方块
回复: 太实用了！
时间: Mon, 02 Jun 2025 08:39:18 +0000

1 view11:14