RISE/Resume-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

aa

Browse Source
This commit is contained in:
RISE
2026-04-27 01:37:22 +08:00
parent ec804afc60
commit e30292e330
2 changed files with 106 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
telegram-scraper.py
View File

@@ -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):