Spaces:
Running
Running
| """ | |
| check_cold_call_desk.py -- read-only: what the Cold Call Desk holds, and where the newest leads land in it. | |
| User report, 2026-09-25: "new leads cold call desk pr nhi dikh rhi, old leads (jin par actions ho chuke) dikh rhi | |
| hain". Uses the SAME rules as the page (crud.list_leads on_desk / DESK_EXCLUDED_STAGES, and the view's grouping | |
| and ordering: New / Follow-ups Due (no_answer) / Voicemail & Hang Ups; New + Follow-ups sort by local-time rank, | |
| then newest first) so what is printed is what the rep sees. | |
| uv run python scripts/check_cold_call_desk.py | |
| Read-only -- changes nothing. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from collections import Counter | |
| from datetime import datetime, timedelta, timezone | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| from sqlalchemy import func, select # noqa: E402 | |
| from app.crud import leads as crud # noqa: E402 | |
| from app.database import SessionLocal # noqa: E402 | |
| from app.models import CallEvent, Lead # noqa: E402 | |
| TRIED = {"voicemail", "hang_up"} | |
| def lane(l: Lead) -> str: | |
| if not l.contact_phone: | |
| return "none" | |
| return {"verified_direct": "direct", "switchboard": "switchboard"}.get(l.phone_status, "none") | |
| def main() -> int: | |
| db = SessionLocal() | |
| try: | |
| desk = list(crud.list_leads(db, desk="cold_call")) | |
| main_list = [l for l in desk if l.last_call_outcome not in TRIED] | |
| print(f"\nOn the desk (sent_to_desk_at set, not hidden, stage not in {crud.DESK_EXCLUDED_STAGES}): {len(desk)}") | |
| print(f" page loads (tried=False): {len(main_list)} Voicemail & Hang Ups (loaded on click): {len(desk) - len(main_list)}") | |
| print(" by last_call_outcome:", dict(Counter(l.last_call_outcome or "(never called)" for l in desk))) | |
| # Leads that already have call actions logged but are still on the desk | |
| acted_ids = set(db.scalars(select(CallEvent.lead_id).where(CallEvent.event_type == "call_outcome")).all()) | |
| acted_on_desk = [l for l in main_list if l.id in acted_ids] | |
| print(f"\nOn the loaded desk list but ALREADY has a logged call outcome: {len(acted_on_desk)}") | |
| for l in acted_on_desk[:15]: | |
| print(f" #{l.id:<5} {l.company_name[:32]:<32} stage={l.pipeline_stage:<14} outcome={l.last_call_outcome or '-':<17} created={str(l.created_at)[:16]}") | |
| # Newest 25 leads by created_at, and whether each is on the desk / where it would sit | |
| newest = list(db.scalars(select(Lead).order_by(Lead.created_at.desc(), Lead.id.desc()).limit(25)).all()) | |
| desk_ids = {l.id for l in desk} | |
| print("\nNewest 25 leads created (any source):") | |
| for l in newest: | |
| on = l.id in desk_ids | |
| why = "" | |
| if not on: | |
| if l.hidden_at: | |
| why = "hidden" | |
| elif not l.sent_to_desk_at: | |
| why = "sent_to_desk_at EMPTY" | |
| elif l.pipeline_stage in crud.DESK_EXCLUDED_STAGES: | |
| why = f"stage {l.pipeline_stage} is not a desk stage" | |
| print( | |
| f" #{l.id:<5} {l.company_name[:30]:<30} created={str(l.created_at)[:16]} desk={'YES' if on else 'no ':<3} " | |
| f"stage={l.pipeline_stage:<13} line={lane(l):<11} tz={l.call_tz if hasattr(l, 'call_tz') else '-'} {why}" | |
| ) | |
| # Position of the newest leads inside the "New" group: rank (line type) beats created_at | |
| new_group = [l for l in main_list if l.last_call_outcome != "no_answer"] | |
| by_lane = Counter(lane(l) for l in new_group) | |
| print(f"\n'New' group size: {len(new_group)} by line type: {dict(by_lane)}") | |
| print(" (the view orders by local-time rank first -- direct/switchboard lines before 'none' -- THEN newest first,") | |
| print(" so a new lead with no usable phone number sorts below every lead that has one.)") | |
| cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=2) | |
| recent = [l for l in new_group if l.created_at and l.created_at >= cutoff] | |
| print(f" created in the last 2 days and in the New group: {len(recent)} by line type: {dict(Counter(lane(l) for l in recent))}") | |
| finally: | |
| db.close() | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |