(self, target_id, target_name, duration_seconds, creds)
| 371 | self.status_data = {} |
| 372 | |
| 373 | async def _monitor_loop(self, target_id, target_name, duration_seconds, creds): |
| 374 | client = TelegramClient(StringSession(creds['tg_session']), int(creds['tg_id']), creds['tg_hash'], loop=get_telethon_loop()) |
| 375 | await client.connect() |
| 376 | if not await client.is_user_authorized(): |
| 377 | self.active_tasks.pop(target_id, None) |
| 378 | return |
| 379 | |
| 380 | end_time = time.time() + duration_seconds |
| 381 | # Clean target_id for filename |
| 382 | safe_id = "".join(c for c in str(target_id) if c.isalnum() or c in ('_', '-')) |
| 383 | log_file = f"TG_Monitor_{safe_id}.txt" |
| 384 | |
| 385 | with open(log_file, "a", encoding="utf-8") as f: |
| 386 | f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] AVVIO MONITORAGGIO per {target_name} ({target_id})\n") |
| 387 | |
| 388 | last_state = None |
| 389 | last_first_name = None |
| 390 | last_last_name = None |
| 391 | last_username = None |
| 392 | |
| 393 | while time.time() < end_time and self.active_tasks.get(target_id): |
| 394 | try: |
| 395 | entity = await client.get_entity(int(target_id)) |
| 396 | is_online = False |
| 397 | last_seen_str = "Sconosciuto" |
| 398 | |
| 399 | if isinstance(entity.status, types.UserStatusOnline): |
| 400 | is_online = True |
| 401 | last_seen_str = "Adesso" |
| 402 | elif getattr(entity.status, "was_online", None): |
| 403 | dt = entity.status.was_online |
| 404 | last_seen_str = dt.strftime("%Y-%m-%d %H:%M:%S") |
| 405 | elif isinstance(entity.status, types.UserStatusRecently): |
| 406 | last_seen_str = "Di recente" |
| 407 | elif isinstance(entity.status, types.UserStatusLastWeek): |
| 408 | last_seen_str = "Entro una settimana" |
| 409 | elif isinstance(entity.status, types.UserStatusLastMonth): |
| 410 | last_seen_str = "Entro un mese" |
| 411 | |
| 412 | self.status_data[target_id] = { |
| 413 | "online": is_online, |
| 414 | "last_seen": last_seen_str, |
| 415 | "first_name": entity.first_name, |
| 416 | "last_name": entity.last_name, |
| 417 | "username": entity.username |
| 418 | } |
| 419 | |
| 420 | # Check for profile changes |
| 421 | changes = [] |
| 422 | if last_first_name is not None and entity.first_name != last_first_name: |
| 423 | changes.append(f"Nome cambiato da '{last_first_name}' a '{entity.first_name}'") |
| 424 | if last_last_name is not None and entity.last_name != last_last_name: |
| 425 | changes.append(f"Cognome cambiato da '{last_last_name}' a '{entity.last_name}'") |
| 426 | if last_username is not None and entity.username != last_username: |
| 427 | changes.append(f"Username cambiato da '{last_username}' a '{entity.username}'") |
| 428 | |
| 429 | if changes: |
| 430 | with open(log_file, "a", encoding="utf-8") as f: |
no test coverage detected