From e30292e33023c50007093b10e9ec40f7d388fadd Mon Sep 17 00:00:00 2001 From: RISE Date: Mon, 27 Apr 2026 01:37:22 +0800 Subject: [PATCH] aa --- app_web.py | 6 ++ telegram-scraper.py | 131 +++++++++++++++++++++++++++++++++----------- 2 files changed, 106 insertions(+), 31 deletions(-) diff --git a/app_web.py b/app_web.py index 390da5f..547f416 100644 --- a/app_web.py +++ b/app_web.py @@ -25,6 +25,9 @@ SCRIPT_FILE = BASE_DIR / "telegram-scraper.py" STATE_FILE = BASE_DIR / "state.json" TEMPLATES_DIR = BASE_DIR / "templates" +# Web 进程：禁用交互式 stdin（避免伪 TTY 下 input() 报 EOF）；代理策略见 telegram-scraper.build_proxy_config +os.environ.setdefault("TELEGRAM_WEB_UI", "1") + logger = logging.getLogger("uvicorn.error") app = FastAPI(title="Telegram Scraper Web Console") @@ -103,6 +106,7 @@ BINARY_ENV_KEYS = frozenset( CONFIG_KEYS: List[Tuple[str, str, str]] = [ ("API_ID", "Telegram API_ID", "text"), ("API_HASH", "Telegram API_HASH", "text"), + ("PROXY_ENABLED", "启用代理（网页端默认关：仅填 1/true/on 才走代理；CLI 本地可留空+填 HOST）", "text"), ("PROXY_TYPE", "代理类型", "text"), ("PROXY_HOST", "代理主机", "text"), ("PROXY_PORT", "代理端口", "text"), @@ -120,6 +124,8 @@ CONFIG_KEYS: List[Tuple[str, str, str]] = [ ("FORWARD_AS_USERNAME", "@替换目标用户名（不带@）", "text"), ("FORWARD_DELAY_SECONDS", "每条推送延迟秒", "text"), ("FORWARD_RAW_MENTIONS", "保留原@不替换（1开/0关）", "text"), + ("TELEGRAM_HEADLESS_QR", "无界面扫码登录（1开：日志里打开二维码链接；服务器/Docker 用）", "text"), + ("TELEGRAM_2FA_PASSWORD", "两步验证密码（仅 headless 扫码需要；勿提交 Git）", "password"), ] diff --git a/telegram-scraper.py b/telegram-scraper.py index 0897de1..d40c665 100644 --- a/telegram-scraper.py +++ b/telegram-scraper.py @@ -375,13 +375,28 @@ class OptimizedTelegramScraper: print(f"\n推送消息 {message.id} 失败（可能无发言权限或媒体无法下载）：{e}") def build_proxy_config(self): - proxy_host = os.getenv("PROXY_HOST") - proxy_port = os.getenv("PROXY_PORT") - if not proxy_host or not proxy_port: + """代理策略： + - TELEGRAM_WEB_UI：网页端默认直连，仅当 PROXY_ENABLED=1/true/on 时才使用代理（避免服务器误配 HOST 即走代理）。 + - 命令行本地：PROXY_ENABLED=0/false/off 禁用；未设置 PROXY_ENABLED 时仍可按 HOST+PORT 启用（兼容旧配置）。 + """ + pe = (os.getenv("PROXY_ENABLED") or "").strip().lower() + web_ui = (os.getenv("TELEGRAM_WEB_UI") or "").strip().lower() in ("1", "true", "yes", "on") + if web_ui: + if pe not in ("1", "true", "yes", "on"): + return None + else: + if pe in ("0", "false", "no", "off", "none"): + return None + proxy_host = (os.getenv("PROXY_HOST") or "").strip() + proxy_port_raw = (os.getenv("PROXY_PORT") or "").strip() + if not proxy_host or not proxy_port_raw: + return None + if not web_ui and pe not in ("", "1", "true", "yes", "on"): + print(f"PROXY_ENABLED 取值无效（{pe!r}），已跳过代理。") return None try: - proxy_port = int(proxy_port) + proxy_port = int(proxy_port_raw) except ValueError: print("环境变量中的 PROXY_PORT 无效，已跳过代理设置。") return None @@ -1045,25 +1060,44 @@ class OptimizedTelegramScraper: f.seek(0) print(f.read()) - async def qr_code_auth(self): + def _stdin_interactive(self) -> bool: + try: + if (os.getenv("TELEGRAM_WEB_UI") or "").strip().lower() in ("1", "true", "yes", "on"): + return False + return sys.stdin is not None and sys.stdin.isatty() + except Exception: + return False + + async def qr_code_auth(self, headless_hint: bool = False): print("\n正在使用二维码登录...") print("请用 Telegram 手机端扫描二维码：") print("1. 在手机上打开 Telegram") print("2. 进入 设置 > 设备 > 扫描二维码") - print("3. 扫描下方二维码\n") - + print("3. 扫描下方二维码（或打开下方链接）\n") + qr_login = await self.client.qr_login() + print(f"二维码链接（可复制到手机浏览器）：\n{qr_login.url}\n") self.display_qr_code_ascii(qr_login) - + if headless_hint: + print("（Web/Docker 环境请查看本页「运行日志」或容器日志中的上述链接。）") + try: await qr_login.wait() print("\n✅ 二维码登录成功！") return True except SessionPasswordNeededError: - password = input("已开启两步验证，请输入密码：") - await self.client.sign_in(password=password) - print("\n✅ 两步验证登录成功！") - return True + pwd = (os.getenv("TELEGRAM_2FA_PASSWORD") or "").strip() + if pwd: + await self.client.sign_in(password=pwd) + print("\n✅ 两步验证登录成功！") + return True + if self._stdin_interactive(): + password = input("已开启两步验证，请输入密码：") + await self.client.sign_in(password=password) + print("\n✅ 两步验证登录成功！") + return True + print("已开启两步验证：请在 .env 中设置 TELEGRAM_2FA_PASSWORD 后重试（勿提交到 Git）。") + return False except Exception as e: print(f"\n❌ 二维码登录失败：{e}") return False @@ -1099,6 +1133,9 @@ class OptimizedTelegramScraper: if not all([self.state.get('api_id'), self.state.get('api_hash')]): print("\n=== 需要配置 API ===") print("请提供来自 https://my.telegram.org 的 API 凭据") + if not self._stdin_interactive(): + print("当前为非交互环境：请在 .env 或网页「环境配置」中填写 API_ID / API_HASH。") + return False try: self.state['api_id'] = int(input("请输入 API ID：")) self.state['api_hash'] = input("请输入 API Hash：") @@ -1106,39 +1143,71 @@ class OptimizedTelegramScraper: except ValueError: print("API ID 无效，必须是数字。") return False + except EOFError: + print("无法读取输入（EOF）。请在 .env 中配置 API_ID / API_HASH。") + return False proxy = self.build_proxy_config() if proxy: print(f"正在使用代理：{proxy[0]}://{proxy[1]}:{proxy[2]}") self.client = TelegramClient('session', self.state['api_id'], self.state['api_hash'], proxy=proxy) - + try: await self.client.connect() except Exception as e: print(f"连接失败：{e}") return False - + if not await self.client.is_user_authorized(): - print("\n=== 请选择登录方式 ===") - print("[1] 二维码登录（推荐，无需手机号）") - print("[2] 手机号登录（传统方式）") - - while True: - choice = input("请输入选项（1 或 2）：").strip() - if choice in ['1', '2']: - break - print("请输入 1 或 2") - - success = await self.qr_code_auth() if choice == '1' else await self.phone_auth() - - if not success: - print("登录失败，请重试。") - await self.client.disconnect() - return False + if not self._stdin_interactive(): + print( + "\n=== Web / Docker 等非交互环境 ===\n" + "无法使用「手机号登录」或控制台选项。\n" + "任选其一：\n" + " A) 将本机已登录生成的 session.session 复制到程序目录（与 app_web.py 同级），再点「初始化/连接」。\n" + " B) 在 .env 设置 TELEGRAM_HEADLESS_QR=1，重启后点「初始化/连接」，在「运行日志」里打开「二维码链接」用手机 Telegram 扫码。\n" + " 若开启了两步验证，可设 TELEGRAM_2FA_PASSWORD（勿提交到 Git）。\n" + ) + headless_qr = (os.getenv("TELEGRAM_HEADLESS_QR") or "").strip().lower() in ( + "1", + "true", + "yes", + "on", + ) + if headless_qr: + success = await self.qr_code_auth(headless_hint=True) + if not success: + await self.client.disconnect() + return False + else: + await self.client.disconnect() + return False + else: + print("\n=== 请选择登录方式 ===") + print("[1] 二维码登录（推荐，无需手机号）") + print("[2] 手机号登录（传统方式）") + + while True: + try: + choice = input("请输入选项（1 或 2）：").strip() + except EOFError: + print("输入结束，已取消登录。") + await self.client.disconnect() + return False + if choice in ["1", "2"]: + break + print("请输入 1 或 2") + + success = await self.qr_code_auth() if choice == "1" else await self.phone_auth() + + if not success: + print("登录失败，请重试。") + await self.client.disconnect() + return False else: print("✅ 已登录！") - + return True def parse_channel_selection(self, choice):