| 1 | from __future__ import annotations |
| 2 | |
| 3 | from dataclasses import dataclass |
| 4 | from uuid import UUID |
| 5 | from urllib.parse import urlparse |
| 6 | |
| 7 | from app.models import DocumentSubtype, ReliabilityLevel, SourceClassification, SourceType |
| 8 | |
| 9 | |
| 10 | @dataclass(frozen=True) |
| 11 | class RegisteredResearchSource: |
| 12 | source_id: str |
| 13 | instrument_id: UUID |
| 14 | url: str |
| 15 | source_type: SourceType |
| 16 | source_name: str |
| 17 | publisher: str |
| 18 | reliability_level: ReliabilityLevel |
| 19 | source_classification: SourceClassification = SourceClassification.OTHER |
| 20 | domain: str = "" |
| 21 | company_id: UUID | None = None |
| 22 | allowed: bool = True |
| 23 | discovery_method: str = "REGISTERED" |
| 24 | priority: int = 1 |
| 25 | categories: tuple[str, ...] = () |
| 26 | # Optional metadata enrichment from NSE announcement discovery. It is not |
| 27 | # part of the source URL or document identity. |
| 28 | document_subtype: DocumentSubtype | None = None |
| 29 | # Set exclusively by OfficialFilingDiscovery after NSE's corporate |
| 30 | # announcements API was queried with this profile's verified NSE symbol. |
| 31 | # It is deliberately not derived from a URL, publisher, or source name. |
| 32 | official_nse_profile_symbol: str | None = None |
| 33 | |
| 34 | @property |
| 35 | def host(self) -> str: |
| 36 | return (self.domain or urlparse(self.url).hostname or "").lower() |
| 37 | |
| 38 | |
| 39 | AIXTRON_INSTRUMENT_ID = UUID("11111111-1111-1111-1111-111111111111") |
| 40 | |
| 41 | |
| 42 | def registered_sources_for(instrument_id: UUID) -> list[RegisteredResearchSource]: |
| 43 | if instrument_id != AIXTRON_INSTRUMENT_ID: |
| 44 | return [] |
| 45 | return [ |
| 46 | RegisteredResearchSource( |
| 47 | source_id="aixtron-official-h1-2026-optoelectronics", |
| 48 | instrument_id=AIXTRON_INSTRUMENT_ID, |
| 49 | url="https://www.aixtron.com/en/press/press-releases/Strong%20momentum%20in%20optoelectronics%20continues_n14146", |
| 50 | source_type=SourceType.INVESTOR_RELATIONS, |
| 51 | source_classification=SourceClassification.OFFICIAL_COMPANY, |
| 52 | source_name="AIXTRON official press release", |
| 53 | publisher="AIXTRON SE", |
| 54 | reliability_level=ReliabilityLevel.LEVEL_B, |
| 55 | domain="www.aixtron.com", |
| 56 | company_id=UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1"), |
| 57 | allowed=True, |
| 58 | discovery_method="REGISTERED_OFFICIAL", |
| 59 | priority=1, |
| 60 | categories=("Orders & Backlog", "CAPEX & Capacity", "Guidance", "Growth", "Customers"), |
| 61 | ) |
| 62 | ] |
| 63 | |
| 64 | |
| 65 | def approved_sources_for_categories(instrument_id: UUID, categories: set[str]) -> list[RegisteredResearchSource]: |
| 66 | sources = [ |
| 67 | source |
| 68 | for source in registered_sources_for(instrument_id) |
| 69 | if source.allowed and (not categories or bool(set(source.categories) & categories)) |
| 70 | ] |
| 71 | return sorted(sources, key=lambda source: (source.priority, source.source_id)) |