forust/telegram-scraper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Activity

Merge branch 'dev'

Browse Source
This commit is contained in:
mr-forust
2026-05-24 21:56:24 +02:00
parent 5b8a95d73d ff000e88d4
commit 6851922a9e
4 changed files with 146 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_scraper_with_forwarding.py
+8 -7
View File
@@ -829,7 +829,7 @@ class OptimizedTelegramScraper:
self.state['channels'][channel] = last_message_id
self.save_state()
print(f"\nCompleted scraping channel {channel}")
print(f"Completed scraping channel {channel}")
except Exception as e:
print(f"Error with channel {channel}: {e}")
@@ -843,10 +843,10 @@ class OptimizedTelegramScraper:
channel_name = self.state.get('channel_names', {}).get(channel, 'Unknown')
if not message_ids:
print(f"No media files to reprocess for {channel_name} (ID: {channel})")
# print(f"No media files to reprocess for {channel_name} (ID: {channel})")
return
print(f"📥 Reprocessing {len(message_ids)} media files for {channel_name} (ID: {channel})")
# print(f"📥 Reprocessing {len(message_ids)} media files for {channel_name} (ID: {channel})")
try:
entity = await self._resolve_entity(channel)
@@ -881,10 +881,10 @@ class OptimizedTelegramScraper:
filled_length = int(bar_length * completed_media // len(message_ids))
bar = '█' * filled_length + '░' * (bar_length - filled_length)
sys.stdout.write(f"\r🔄 Rescrape: [{bar}] {progress:.1f}% ({completed_media}/{len(message_ids)})")
sys.stdout.flush()
# sys.stdout.write(f"\r🔄 Rescrape: [{bar}] {progress:.1f}% ({completed_media}/{len(message_ids)})")
# sys.stdout.flush()
print(f"\n✅ Media reprocessing complete! ({successful_downloads}/{len(message_ids)} successful)")
# print(f"\n✅ Media reprocessing complete! ({successful_downloads}/{len(message_ids)} successful)")
except Exception as e:
print(f"Error reprocessing media: {e}")
@@ -1306,7 +1306,8 @@ class OptimizedTelegramScraper:
await self.client.disconnect()
return False
else:
print("✅ Already authenticated!")
pass
# print("✅ Already authenticated!")
return True
webui/app.js
+8
View File
@@ -84,6 +84,14 @@ function renderChannels(channels) {
await refreshDashboard();
});
node.querySelector(".export-btn").addEventListener("click", async () => {
await api("/api/jobs/export-channel", {
method: "POST",
body: JSON.stringify({ channel_id: channel.channel_id }),
});
await refreshDashboard();
});
node.querySelector(".export-view-btn").addEventListener("click", () => {
window.location.href = `/viewer?channel=${encodeURIComponent(channel.channel_id)}`;
});
webui/index.html
+1 -2
View File
@@ -194,8 +194,7 @@
<td class="last-date"></td>
<td>
<div class="action-row">
<button class="button button-small scrape-btn">Scrape</button>
<button class="button button-small export-view-btn">Viewer</button>
<button class="button button-small scrape-btn">Scrape</button> <button class="button button-small export-btn">Export</button> <button class="button button-small export-view-btn">Viewer</button>
<button class="button button-small media-btn">Rescrape media</button>
<button class="button button-small danger remove-btn">Удалить</button>
</div>
webui_server.py
+129 -35
View File
@@ -63,7 +63,9 @@ def channel_db_path(channel_id: str) -> Path:
return DATA_DIR / channel_id / f"{channel_id}.db"
def guess_media_kind(media_path: Optional[str], media_type: Optional[str]) -> Optional[str]:
def guess_media_kind(
media_path: Optional[str], media_type: Optional[str]
) -> Optional[str]:
if not media_path:
return None
suffix = Path(media_path).suffix.lower()
@@ -129,10 +131,7 @@ def database_summary(channel_id: str) -> Dict[str, Any]:
def channel_display_name(state: Dict[str, Any], channel_id: str) -> str:
return (
state.get("channel_names", {}).get(channel_id)
or channel_id
)
return state.get("channel_names", {}).get(channel_id) or channel_id
def list_channels_snapshot() -> List[Dict[str, Any]]:
@@ -158,7 +157,9 @@ def list_channels_snapshot() -> List[Dict[str, Any]]:
return channels
def load_messages(channel_id: str, limit: int = 120, before_message_id: Optional[int] = None) -> List[Dict[str, Any]]:
def load_messages(
channel_id: str, limit: int = 120, before_message_id: Optional[int] = None
) -> List[Dict[str, Any]]:
db_path = channel_db_path(channel_id)
if not db_path.exists():
return []
@@ -404,7 +405,11 @@ class TelegramAuthManager:
async def _start_qr_login(self) -> Dict[str, Any]:
client = await self._get_client()
if await client.is_user_authorized():
self._set_state(phase="authorized", status="authorized", details="Telegram session is already authorized.")
self._set_state(
phase="authorized",
status="authorized",
details="Telegram session is already authorized.",
)
return self.snapshot()
self.qr_login = await client.qr_login()
qr_url = self.qr_login.url
@@ -424,7 +429,11 @@ class TelegramAuthManager:
async def _request_phone_code(self, phone: str) -> Dict[str, Any]:
client = await self._get_client()
if await client.is_user_authorized():
self._set_state(phase="authorized", status="authorized", details="Telegram session is already authorized.")
self._set_state(
phase="authorized",
status="authorized",
details="Telegram session is already authorized.",
)
return self.snapshot()
sent = await client.send_code_request(phone)
self.phone = phone
@@ -445,7 +454,9 @@ class TelegramAuthManager:
if not self.phone or not self.phone_code_hash:
raise RuntimeError("Request a phone code first.")
try:
await client.sign_in(phone=self.phone, code=code, phone_code_hash=self.phone_code_hash)
await client.sign_in(
phone=self.phone, code=code, phone_code_hash=self.phone_code_hash
)
self._set_state(
phase="authorized",
status="authorized",
@@ -518,7 +529,13 @@ class ContinuousScrapeManager:
},
}
def update(self, enabled: bool, interval_minutes: int, channels: List[str], run_all_tracked: bool) -> Dict[str, Any]:
def update(
self,
enabled: bool,
interval_minutes: int,
channels: List[str],
run_all_tracked: bool,
) -> Dict[str, Any]:
interval_minutes = max(1, int(interval_minutes))
with self.lock:
self.config = {
@@ -572,10 +589,15 @@ class ContinuousScrapeManager:
if not channels:
self._log("No channels configured for continuous scraping.")
else:
self._log(f"Starting iteration for {len(channels)} channel(s).")
self._log("=" * 50)
self._log(f"🚀 Starting iteration for {len(channels)} channel(s).")
self._log("-" * 50)
buffer = io.StringIO()
try:
with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer):
with (
contextlib.redirect_stdout(buffer),
contextlib.redirect_stderr(buffer),
):
run_job("scrape_selected", {"channels": channels})
output = buffer.getvalue().strip()
if output:
@@ -585,7 +607,8 @@ class ContinuousScrapeManager:
self.status["last_iteration_at"] = utc_now_iso()
self.status["last_finished_at"] = utc_now_iso()
self.status["last_error"] = None
self._log("Iteration finished.")
self._log("-" * 50)
self._log("✅ Iteration finished.")
except Exception as exc:
output = buffer.getvalue().strip()
if output:
@@ -596,7 +619,7 @@ class ContinuousScrapeManager:
self.status["last_error"] = str(exc)
sleep_seconds = max(5, interval_minutes * 60)
self._log(f"Sleeping for {interval_minutes} minute(s).")
self._log(f"💤 Sleeping for {interval_minutes} minute(s).")
interrupted = self.stop_event.wait(timeout=sleep_seconds)
if interrupted:
break
@@ -637,15 +660,25 @@ def run_job(job_type: str, payload: Dict[str, Any]) -> None:
elif job_type == "scrape_all":
channels = list(scraper.state.get("channels", {}).keys())
for channel_id in channels:
offset = int(scraper.state.get("channels", {}).get(channel_id, 0) or 0)
offset = int(
scraper.state.get("channels", {}).get(channel_id, 0) or 0
)
await scraper.scrape_channel(channel_id, offset)
elif job_type == "scrape_selected":
channels = [str(channel_id) for channel_id in payload.get("channels", [])]
channels = [
str(channel_id) for channel_id in payload.get("channels", [])
]
for channel_id in channels:
offset = int(scraper.state.get("channels", {}).get(channel_id, 0) or 0)
offset = int(
scraper.state.get("channels", {}).get(channel_id, 0) or 0
)
await scraper.scrape_channel(channel_id, offset)
elif job_type == "export_all":
await scraper.export_data()
elif job_type == "export_channel":
channel_id = payload["channel_id"]
scraper.export_to_csv(channel_id)
scraper.export_to_json(channel_id)
elif job_type == "rescrape_media":
await scraper.rescrape_media(payload["channel_id"])
elif job_type == "fix_missing_media":
@@ -682,7 +715,9 @@ def auth_status() -> Dict[str, Any]:
status["session_ready"] = session_file.exists()
if status["has_api_credentials"] and status["session_ready"]:
status["status"] = "ready"
status["details"] = "Scraping actions should be available after dependencies are installed."
status["details"] = (
"Scraping actions should be available after dependencies are installed."
)
elif status["has_api_credentials"]:
status["status"] = "needs_auth"
status["details"] = "API keys found, but Telegram session file is missing."
@@ -722,7 +757,9 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
if path == "/":
return self.serve_file(WEBUI_DIR / "index.html", "text/html; charset=utf-8")
if path == "/viewer":
return self.serve_file(WEBUI_DIR / "viewer.html", "text/html; charset=utf-8")
return self.serve_file(
WEBUI_DIR / "viewer.html", "text/html; charset=utf-8"
)
if path.startswith("/static/"):
relative = path.removeprefix("/static/")
return self.serve_static(relative)
@@ -735,6 +772,8 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
return self.send_json(self.app.auth_manager.snapshot())
if path == "/api/continuous":
return self.send_json(self.app.continuous_manager.snapshot())
if path == "/health/continuous":
return self.send_json(self.app.continuous_manager.snapshot())
if path == "/api/jobs":
return self.send_json(self.app.job_runner.recent_jobs())
if path.startswith("/api/jobs/"):
@@ -753,9 +792,15 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
before_message_id = int(before[0]) if before else None
payload = {
"channel_id": channel_id,
"messages": load_messages(channel_id, limit=limit, before_message_id=before_message_id),
"messages": load_messages(
channel_id, limit=limit, before_message_id=before_message_id
),
"channel": next(
(item for item in list_channels_snapshot() if item["channel_id"] == channel_id),
(
item
for item in list_channels_snapshot()
if item["channel_id"] == channel_id
),
None,
),
}
@@ -767,17 +812,31 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
path = parsed.path
if path == "/":
return self.serve_file(WEBUI_DIR / "index.html", "text/html; charset=utf-8", head_only=True)
return self.serve_file(
WEBUI_DIR / "index.html", "text/html; charset=utf-8", head_only=True
)
if path == "/viewer":
return self.serve_file(WEBUI_DIR / "viewer.html", "text/html; charset=utf-8", head_only=True)
return self.serve_file(
WEBUI_DIR / "viewer.html", "text/html; charset=utf-8", head_only=True
)
if path.startswith("/static/"):
relative = path.removeprefix("/static/")
return self.serve_static(relative, head_only=True)
if path.startswith("/media/"):
relative = path.removeprefix("/media/")
return self.serve_media(relative, head_only=True)
if path in {"/api/dashboard", "/api/auth", "/api/continuous", "/api/jobs", "/api/channels"} or path.startswith("/api/jobs/") or (
path.startswith("/api/channels/") and path.endswith("/messages")
if (
path
in {
"/api/dashboard",
"/api/auth",
"/api/continuous",
"/api/jobs",
"/api/channels",
"/health/continuous",
}
or path.startswith("/api/jobs/")
or (path.startswith("/api/channels/") and path.endswith("/messages"))
):
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "application/json; charset=utf-8")
@@ -805,7 +864,11 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
try:
enabled = bool(body.get("enabled"))
interval_minutes = int(body.get("interval_minutes", 1))
channels = [str(item).strip() for item in body.get("channels", []) if str(item).strip()]
channels = [
str(item).strip()
for item in body.get("channels", [])
if str(item).strip()
]
run_all_tracked = bool(body.get("run_all_tracked", True))
payload = self.app.continuous_manager.update(
enabled=enabled,
@@ -821,7 +884,9 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
api_id = body.get("api_id")
api_hash = str(body.get("api_hash", "")).strip()
if not api_id or not api_hash:
return self.send_error_json(HTTPStatus.BAD_REQUEST, "api_id and api_hash are required")
return self.send_error_json(
HTTPStatus.BAD_REQUEST, "api_id and api_hash are required"
)
try:
payload = self.app.auth_manager.save_credentials(int(api_id), api_hash)
except Exception as exc:
@@ -858,7 +923,9 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
if path == "/api/auth/password":
password = str(body.get("password", "")).strip()
if not password:
return self.send_error_json(HTTPStatus.BAD_REQUEST, "password is required")
return self.send_error_json(
HTTPStatus.BAD_REQUEST, "password is required"
)
try:
payload = self.app.auth_manager.submit_password(password)
except Exception as exc:
@@ -868,12 +935,16 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
if path == "/api/channels/add":
channel_id = str(body.get("channel_id", "")).strip()
if not channel_id:
return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required")
return self.send_error_json(
HTTPStatus.BAD_REQUEST, "channel_id is required"
)
state = load_state()
if channel_id not in state["channels"]:
state["channels"][channel_id] = 0
if body.get("name"):
state.setdefault("channel_names", {})[channel_id] = str(body["name"]).strip()
state.setdefault("channel_names", {})[channel_id] = str(
body["name"]
).strip()
save_state(state)
return self.send_json({"ok": True, "channel_id": channel_id})
@@ -909,10 +980,25 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
)
return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)
if path == "/api/jobs/export-channel":
channel_id = str(body.get("channel_id", "")).strip()
if not channel_id:
return self.send_error_json(
HTTPStatus.BAD_REQUEST, "channel_id is required"
)
job = self.app.job_runner.create_job(
"export_channel",
f"Export channel {channel_id}",
{"channel_id": channel_id},
)
return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)
if path == "/api/jobs/rescrape-media":
channel_id = str(body.get("channel_id", "")).strip()
if not channel_id:
return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required")
return self.send_error_json(
HTTPStatus.BAD_REQUEST, "channel_id is required"
)
job = self.app.job_runner.create_job(
"rescrape_media",
f"Rescrape media for {channel_id}",
@@ -923,7 +1009,9 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
if path == "/api/jobs/fix-media":
channel_id = str(body.get("channel_id", "")).strip()
if not channel_id:
return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required")
return self.send_error_json(
HTTPStatus.BAD_REQUEST, "channel_id is required"
)
job = self.app.job_runner.create_job(
"fix_missing_media",
f"Fix missing media for {channel_id}",
@@ -959,7 +1047,9 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
return self.send_error_json(HTTPStatus.FORBIDDEN, "Forbidden")
if not file_path.exists() or not file_path.is_file():
return self.send_error_json(HTTPStatus.NOT_FOUND, "File not found")
mime_type = mimetypes.guess_type(str(file_path))[0] or "application/octet-stream"
mime_type = (
mimetypes.guess_type(str(file_path))[0] or "application/octet-stream"
)
return self.serve_file(file_path, mime_type, head_only=head_only)
def serve_media(self, relative_path: str, head_only: bool = False) -> None:
@@ -971,10 +1061,14 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
return self.send_error_json(HTTPStatus.FORBIDDEN, "Forbidden")
if not file_path.exists() or not file_path.is_file():
return self.send_error_json(HTTPStatus.NOT_FOUND, "Media file not found")
mime_type = mimetypes.guess_type(str(file_path))[0] or "application/octet-stream"
mime_type = (
mimetypes.guess_type(str(file_path))[0] or "application/octet-stream"
)
return self.serve_file(file_path, mime_type, head_only=head_only)
def serve_file(self, file_path: Path, content_type: str, head_only: bool = False) -> None:
def serve_file(
self, file_path: Path, content_type: str, head_only: bool = False
) -> None:
data = file_path.read_bytes()
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", content_type)