| 1 | """Explicit bounded news worker. Never imported/called by ranking or GET flows.""" |
| 2 | import asyncio |
| 3 | from datetime import datetime, timezone, timedelta |
| 4 | from uuid import NAMESPACE_URL, uuid5 |
| 5 | from urllib.parse import urlparse |
| 6 | from app.business_exposure import BusinessEvidence, SourceReference, extract_profile, query_plan |
| 7 | from app.news_intelligence import ProviderOutcome, aggregate_search, extract_impacts |
| 8 | from app.source_discovery import SearchDateWindow, CandidateSearchResult, classify_source, reliability_for_classification, source_type_for_classification |
| 9 | from app.normalization import canonicalize_url, content_hash, extract_text, extract_published_at |
| 10 | from app.models import ResearchDocument, SourceMode |
| 11 | |
| 12 | |
| 13 | def persisted_yahoo_discovery(repository, company, now): |
| 14 | """Reuse an explicitly recorded Yahoo news outcome; never infer search |
| 15 | success from an empty general-purpose quote snapshot. One company-news |
| 16 | check is supplemental to, and cannot replace, the web query plan. |
| 17 | """ |
| 18 | loader=getattr(repository,'acquisition_observations_for',None) |
| 19 | checks=[] |
| 20 | for row in loader(company.instrument_id) if callable(loader) else []: |
| 21 | if row.get('requirement_id')!='CURRENT_NEWS' or row.get('provider')!='YAHOO_FINANCE_MCP': continue |
| 22 | stamp=row.get('observed_at') |
| 23 | if isinstance(stamp,str): stamp=datetime.fromisoformat(stamp.replace('Z','+00:00')) |
| 24 | if stamp and stamp.tzinfo and timedelta(0)<=now-stamp<=timedelta(days=1): checks.append((stamp,row)) |
| 25 | if not checks: return [],[] |
| 26 | stamp,row=max(checks,key=lambda item:item[0]) |
| 27 | candidates=[] |
| 28 | success=row.get('outcome') in {'SUCCESS','SUCCESS_EMPTY'} |
| 29 | if row.get('outcome')=='SUCCESS': |
| 30 | for doc in repository.documents_for(company.instrument_id,source_mode=SourceMode.REAL): |
| 31 | if doc.discovery_provider=='YAHOO_FINANCE_MCP' and doc.retrieved_at<=stamp: |
| 32 | candidates.append(CandidateSearchResult(doc.title,doc.canonical_url,'',doc.discovered_at or doc.retrieved_at, |
| 33 | 'YAHOO_FINANCE_MCP','persisted-company-news','persisted company news','CURRENT_NEWS')) |
| 34 | state=('SUCCESS_WITH_RESULTS' if candidates else 'SUCCESS_EMPTY') if success else 'FAILED' |
| 35 | if row.get('outcome')=='SUCCESS' and not candidates: state='PARTIAL' |
| 36 | return [ProviderOutcome(provider='YAHOO_FINANCE_MCP',outcome=state,candidate_count=len(candidates), |
| 37 | queries_planned=1,queries_completed=int(success),failure_code=None if success else 'PERSISTED_YAHOO_SEARCH_FAILED')],candidates |
| 38 | |
| 39 | def persisted_business_evidence(repository, company): |
| 40 | evidence=[] |
| 41 | for doc in repository.documents_for(company.instrument_id,source_mode=SourceMode.REAL): |
| 42 | if doc.normalized_text and doc.source_classification in {'OFFICIAL_COMPANY','EXCHANGE','REGULATORY','COMPANY_FILING'}: |
| 43 | evidence.append(BusinessEvidence(instrument_id=company.instrument_id,text=doc.normalized_text, |
| 44 | source=SourceReference(document_id=doc.document_id,url=doc.canonical_url,classification=doc.source_classification, |
| 45 | publication_time=doc.published_at,retrieved_at=doc.retrieved_at,confidence=.95))) |
| 46 | snapshots=repository.structured_market_snapshots_for({company.instrument_id}).get(company.instrument_id,[]) |
| 47 | for record in snapshots: |
| 48 | description=record.snapshot.facts.get('businessSummary') |
| 49 | if description and description.value and record.provider in {'YAHOO_FINANCE','NSE','BSE'}: |
| 50 | evidence.append(BusinessEvidence(instrument_id=company.instrument_id,text=str(description.value), |
| 51 | source=SourceReference(url=description.source_url,classification='STRUCTURED_MARKET_PROVIDER', |
| 52 | publication_time=description.published_at,retrieved_at=description.retrieved_at,confidence=description.confidence or 0))) |
| 53 | return evidence |
| 54 | |
| 55 | async def acquire_news(repository, company, *, providers, industry=None, now=None, max_queries=14, max_documents=6, sleep=asyncio.sleep): |
| 56 | """One company; injected existing search adapters and existing safe fetcher. |
| 57 | |
| 58 | A document budget exhaustion or failed fetch makes the run PARTIAL; it can |
| 59 | never certify no-events. Empty results certify only the configured plan. |
| 60 | """ |
| 61 | if not 1<=len(providers)<=3 or not 1<=max_queries<=20 or not 1<=max_documents<=20: raise ValueError('INVALID_NEWS_BUDGET') |
| 62 | started=now or datetime.now(timezone.utc) |
| 63 | if industry is None: |
| 64 | records=repository.structured_market_snapshots_for({company.instrument_id}).get(company.instrument_id,[]) |
| 65 | for record in sorted(records,key=lambda r:r.retrieved_at,reverse=True): |
| 66 | fact=record.snapshot.facts.get('industry') |
| 67 | if fact and fact.value and fact.retrieved_at<=started: |
| 68 | industry=str(fact.value); break |
| 69 | exposure=extract_profile(company.instrument_id,company.company_name,company.ticker, |
| 70 | persisted_business_evidence(repository,company),now=started,industry=industry) |
| 71 | exposure=await repository.append_news_record(exposure) |
| 72 | plan=query_plan(exposure) |
| 73 | # Fair tier allocation under the platform's smaller operational budget. |
| 74 | buckets=[[q for q in plan if q[0]==tier] for tier in (1,2,3,4)] |
| 75 | plan=[b[i] for i in range(max(map(len,buckets),default=0)) for b in buckets if i<len(b)][:max_queries] |
| 76 | outcomes,supplemental=persisted_yahoo_discovery(repository,company,started) |
| 77 | candidates={canonicalize_url(row.url):row for row in supplemental} |
| 78 | for provider in sorted(providers,key=lambda p:p.provider_name): |
| 79 | if provider.provider_name.lower() == 'disabled': |
| 80 | outcomes.append(ProviderOutcome(provider=provider.provider_name,outcome='FAILED',candidate_count=0, |
| 81 | queries_planned=len(plan),queries_completed=0,failure_code='SEARCH_PROVIDER_DISABLED')) |
| 82 | continue |
| 83 | completed=0; count=0; failed=False; degraded=False |
| 84 | for _,query in plan: |
| 85 | try: |
| 86 | rows=await provider.discover(company,'CURRENT_NEWS',SearchDateWindow(query_limit=1,explicit_queries=(query,))) |
| 87 | completed+=1; count+=len(rows) |
| 88 | degraded |= bool(getattr(provider,'last_query_degraded',False)) |
| 89 | for row in rows: candidates.setdefault(canonicalize_url(row.url),row) |
| 90 | except Exception: |
| 91 | failed=True |
| 92 | await sleep(repository.settings.market_data_population_request_interval_seconds) |
| 93 | state=('PARTIAL' if failed or degraded else 'SUCCESS_WITH_RESULTS' if count else 'SUCCESS_EMPTY') if completed else 'FAILED' |
| 94 | outcomes.append(ProviderOutcome(provider=provider.provider_name,outcome=state,candidate_count=count, |
| 95 | queries_planned=len(plan),queries_completed=completed,failure_code='SEARCH_PROVIDER_UNAVAILABLE' if failed else None)) |
| 96 | features=[]; incomplete=len(candidates)>max_documents |
| 97 | for url,candidate in sorted(candidates.items())[:max_documents]: |
| 98 | try: |
| 99 | existing=next((d for d in repository.documents.values() if d.canonical_url==url),None) |
| 100 | if existing: |
| 101 | document=existing |
| 102 | else: |
| 103 | fetched=await repository._fetcher.fetch(url) # Existing redirects/SSRF/robots/size policy. |
| 104 | title,body=extract_text(fetched.text,fetched.content_type) |
| 105 | if not body: raise ValueError('EMPTY_DOCUMENT') |
| 106 | classification=classify_source(urlparse(fetched.final_url).hostname or '',company,candidate) |
| 107 | stamp=now or datetime.now(timezone.utc) |
| 108 | document=ResearchDocument(document_id=uuid5(NAMESPACE_URL,url),canonical_url=url,original_url=url, |
| 109 | title=title or candidate.title,source_type=source_type_for_classification(classification),source_classification=classification, |
| 110 | source_name=candidate.provider,content_type=fetched.content_type,document_type='HTML', |
| 111 | normalized_text=body,content_hash=content_hash(body),instrument_id=company.instrument_id,company_id=company.company_id, |
| 112 | reliability_level=reliability_for_classification(classification),source_mode='REAL',status='PARSED', |
| 113 | retrieved_at=stamp,discovered_at=candidate.discovered_at, |
| 114 | published_at=extract_published_at(fetched.text),discovery_provider=candidate.provider) |
| 115 | await repository._run_blocking_persistence(repository._persistence.upsert_document,document) |
| 116 | repository.documents[document.document_id]=document |
| 117 | if not document.normalized_text: raise ValueError('DOCUMENT_BODY_UNAVAILABLE') |
| 118 | extracted=extract_impacts(exposure,document,now=now or datetime.now(timezone.utc)) |
| 119 | for feature in extracted: |
| 120 | features.append(await repository.append_news_record(feature)) |
| 121 | except Exception: |
| 122 | incomplete=True |
| 123 | if incomplete: |
| 124 | outcomes=[p.model_copy(update={'outcome':'PARTIAL'}) if p.outcome.startswith('SUCCESS') else p for p in outcomes] |
| 125 | run=aggregate_search(company.instrument_id,outcomes,started_at=started,completed_at=now or datetime.now(timezone.utc), |
| 126 | qualifying_events=len({f.event_key for f in features}),query_plan=plan) |
| 127 | await repository.append_news_record(run) |
| 128 | return run,features |