""" Heuristics inspired by civic meeting scrapers (e.g. City-Bureau city-scrapers_): detect common vendor stacks and extract document / navigation URLs — **without** Scrapy. .. _city-scrapers: https://github.com/City-Bureau/city-scrapers """ from __future__ import annotations import re from html import unescape from typing import Any, Dict, FrozenSet, List, Optional, Sequence, Set, Tuple from urllib.parse import parse_qs, quote_plus, urljoin, urlparse, urlunparse from bs4 import BeautifulSoup from loguru import logger def _beautifulsoup_meetings(html: Optional[str], *, page_url: str, log_label: str) -> Optional[BeautifulSoup]: """ Parse page HTML for meeting heuristics. Returns ``None`` when the body is empty or when BeautifulSoup rejects the markup (e.g. gzipped/binary served as ``text/html``, corrupted CDATA) so crawls keep going. """ raw = html or "" if not raw.strip(): return None try: return BeautifulSoup(raw, "html.parser") except Exception as exc: logger.debug( "meetings_bs4_skip label={lbl} page_url={pu!r} exc_type={tn} exc={ex!r}", lbl=log_label, pu=page_url, tn=type(exc).__name__, ex=exc, ) return None # Hosts where PDFs / meeting UI often live off the jurisdiction’s marketing domain. _OFFSITE_SUFFIXES: Tuple[str, ...] = ( "legistar.com", "legistar1.granicus.com", "legistarcloud.com", "granicus.com", "granicusideas.com", "civicclerk.com", "civicweb.net", "revize.com", "civicplus.com", "streamlinevillage.com", "boarddocs.com", "municodemeetings.com", "suiteonemedia.com", "eboardsolutions.com", # PowerDMS public libraries (agenda/minutes archives), e.g. Tuscaloosa County Commission. "powerdms.com", ) # Offsite HTML we may follow (narrow path hints to avoid crawling the whole vendor CDN). _VENDOR_PATH_SNIPPETS: Tuple[str, ...] = ( "viewmeeting", "viewpublisher", "mediaplayer", "calendar.aspx", "meetingdetail", "meetinginformation", "/portal/", "view.ashx", "legistar", "granicus", "events.aspx", "event.aspx", "commission", "cityclerk", ) _STACK_PATTERNS: Tuple[Tuple[str, re.Pattern], ...] = ( ("youtube", re.compile(r"youtube\.com|youtu\.be|youtube-nocookie\.com", re.I)), ("legistar", re.compile(r"legistar\.(com|net)|/legistar/|view\.ashx", re.I)), ("granicus", re.compile(r"granicus\.com|granicusideas\.com|viewmeeting\.aspx|viewpublisher", re.I)), ("civicclerk", re.compile(r"civicclerk\.com", re.I)), ("civicweb", re.compile(r"civicweb\.net", re.I)), ("revize", re.compile(r"revize\.com", re.I)), ("wordpress", re.compile(r"/wp-content/|/wp-json/|wordpress|xmlrpc\.php", re.I)), ("boarddocs", re.compile(r"boarddocs\.com", re.I)), ("powerdms", re.compile(r"powerdms\.com", re.I)), ("eboard_simbli", re.compile(r"eboardsolutions\.com|simbli\.eboardsolutions", re.I)), ) _DOC_TYPE_RULES: Tuple[Tuple[str, re.Pattern], ...] = ( ( "minutes", re.compile( r"minute|approved\s*minute|action\s*minutes?|proceedings", re.I, ), ), ("agenda", re.compile(r"agenda|packet", re.I)), ("transcript", re.compile(r"transcript|caption|verbatim", re.I)), ("video", re.compile(r"video|webcast|recording|youtube|vimeo", re.I)), ) def _host(url: str) -> str: try: return urlparse(url).netloc.lower() except Exception: return "" def _host_matches_suffix(host: str, suffix: str) -> bool: h = host.rstrip(".").lower() s = suffix.lower().lstrip(".") return h == s or h.endswith("." + s) def _link_anchor_text(tag: Any) -> str: """Visible link text; Wix often leaves ```` empty and uses ``aria-label``.""" text = (tag.get_text() or "").strip() if text: return text for attr in ("aria-label", "title", "data-label"): val = (tag.get(attr) or "").strip() if val: return val return "" def is_trusted_offsite(url: str) -> bool: h = _host(url) if not h: return False return any(_host_matches_suffix(h, s) for s in _OFFSITE_SUFFIXES) def is_amazonaws_document_cdn_host(url: str) -> bool: """ True when the URL host is an AWS object store / CDN hostname (``*.amazonaws.com``). Used only together with :data:`MEETING_DOWNLOAD_EXT` so we do not crawl arbitrary HTML off S3. """ h = _host(url) return bool(h) and "amazonaws.com" in h def _host_without_www(netloc: str) -> str: """Compare apex and ``www`` as the same site (Revize / county sites often mix both).""" h = (netloc or "").lower().split(":")[0].rstrip(".") return h[4:] if h.startswith("www.") else h def is_same_site(url: str, homepage: str) -> bool: try: a = urlparse(url).netloc b = urlparse(homepage).netloc if not a or not b: return False return _host_without_www(a) == _host_without_www(b) except Exception: return False def _path_query_lower(url: str) -> str: try: p = urlparse(url) return f"{p.path.lower()}?{p.query.lower()}" except Exception: return (url or "").lower() def is_suiteone_document_download_url(url: str) -> bool: """SuiteOne agenda/minutes endpoints without a ``.pdf`` suffix (response is still PDF).""" try: host = urlparse(url).netloc.lower().split(":")[0].rstrip(".") except Exception: return False if not host or not _host_matches_suffix(host, "suiteonemedia.com"): return False pq = _path_query_lower(url) return "/event/getagendafile/" in pq or "/event/getminutesfile/" in pq def is_vendor_meeting_page_url(url: str) -> bool: pq = _path_query_lower(url) try: host = urlparse(url).netloc.lower().split(":")[0].rstrip(".") except Exception: host = "" if host and _host_matches_suffix(host, "suiteonemedia.com"): if "/web/home.aspx" in pq or "/event/" in pq or "embed=1" in pq: return True if host and _host_matches_suffix(host, "eboardsolutions.com"): if "/sb_meetings/" in pq or "/sb_online" in pq.replace("\\", "/"): return True if any( tok in pq for tok in ( "meetinglisting.aspx", "meeting.aspx", "agenda.aspx", "minutes.aspx", "onlineagenda", "onlineminutes", "viewagenda", "viewminutes", ) ): return True if host and _host_matches_suffix(host, "powerdms.com"): if re.search(r"(login|oauth|authorize|logout|sso)(/|\?|$)", pq, re.I): return False # Public libraries / document folders linked from county agenda index pages. if re.search( r"(/libraries?/|/documents?/|/files?/|/browse|/share|/published" r"|documentid=|fileid=|libraryid=|folderid=|recordid=|permalink" r"|commission|agenda|minute|meeting|archive|prior|published)", pq, re.I, ): return True return False if not is_trusted_offsite(url): return False return any(snippet in pq for snippet in _VENDOR_PATH_SNIPPETS) def is_simbli_eboard_host(url: str) -> bool: h = _host(url) return bool(h and _host_matches_suffix(h, "eboardsolutions.com")) def is_simbli_meeting_listing_page_url(url: str) -> bool: """True on Simbli meeting index pages that expand ``ViewMeeting`` rows into crawl seeds.""" if not is_simbli_eboard_host(url): return False pq = _path_query_lower(url) return "meetinglisting.aspx" in pq or "sb_meetinglisting" in pq _SIMBLI_SKIP_HTML_PRINT_MARKERS: Tuple[str, ...] = ( "meetinglisting.aspx", "sb_meetinglisting", "login.aspx", "logout.aspx", "register.aspx", "membership.aspx", "directory.aspx", "changepassword.aspx", ) def is_simbli_agenda_minutes_html_target(url: str, anchor_text: str) -> bool: """ Simbli / eBoard Solutions often serves agendas and minutes as **ASP.NET HTML pages**, not PDFs. Conservative: only URLs that look like agenda/minutes viewers (plus anchor/text classified as agenda/minutes on allowed paths). """ if not is_simbli_eboard_host(url): return False low = _path_query_lower(url) if any(m in low for m in _SIMBLI_SKIP_HTML_PRINT_MARKERS): return False try: path_only = urlparse(url).path.lower() except Exception: path_only = "" blob_l = f"{low} {path_only}".lower() ct = classify_document(url, anchor_text) if ct in ("agenda", "minutes"): # Still exclude navigation shells. if "meetinglisting.aspx" in blob_l: return False return True # URL-shaped agenda/minutes pages even when anchor text is generic ("View", "Online"). if re.search( r"(online)?agenda|viewagenda|/agenda\.aspx|agenda_|_agenda|aid=", blob_l, re.I, ): return True if re.search( r"(online)?minutes|viewminutes|/minutes\.aspx|minutes_|_minutes|mid=", blob_l, re.I, ): return True # Meeting hub opened from SB_MeetingListing grids (onclick ``ViewMeeting`` → ``ViewMeeting.aspx``). if "viewmeeting.aspx" in blob_l and "mid=" in blob_l: return True return False _VIEWMEETING_ONCLICK_RE = re.compile( r"""ViewMeeting\s*\(\s*(?:["']|")(\d+)(?:["']|")\s*,\s*(?:["']|")(\d+)(?:["']|")""", re.I, ) _VIEWMINUTES_ONCLICK_RE = re.compile( r"""ViewMinutes\s*\(\s*(?:["']|")(\d+)(?:["']|")\s*,\s*(?:["']|")(\d+)(?:["']|")\s*,\s*(?:["']|")([^"']+)(?:["']|")""", re.I, ) def extract_simbli_meeting_listing_synthetic_nav_urls(html: str, page_url: str) -> List[Tuple[str, str]]: """ Simbli ``SB_MeetingListing.aspx`` rows often use ``href="javascript:void(0)"`` with ``onclick`` calling ``ViewMeeting`` / ``ViewMinutes``. Expand those into real ``ViewMeeting.aspx`` URLs so the crawl can reach agendas/minutes (see ``SB_MeetingListingJS.js``). Note: Listing grids often load **50 meetings per chunk** via AJAX — only meetings present in the supplied HTML can be expanded here (additional pages may require scrolling/pagination in-browser). """ if not html or not is_simbli_eboard_host(page_url): return [] low = (html or "").lower() if "meetinglisting.aspx" not in low and "sb_meetinglisting" not in low: return [] if "viewmeeting(" not in low: return [] soup = _beautifulsoup_meetings(html, page_url=page_url, log_label="extract_simbli_listing_syn") if soup is None: return [] try: base_root = f"{urlparse(page_url).scheme}://{urlparse(page_url).netloc}" except Exception: return [] seen: Set[str] = set() out: List[Tuple[str, str]] = [] def push(abs_url: str, label: str) -> None: nu = urlunparse(urlparse(abs_url)._replace(fragment="")) if nu not in seen: seen.add(nu) out.append((nu, label.strip()[:500])) for tag in soup.find_all(attrs={"onclick": True}): oc_raw = str(tag.get("onclick") or "") oc = unescape(oc_raw).replace(""", '"') vm = _VIEWMEETING_ONCLICK_RE.search(oc) if vm: sid, mid = vm.group(1), vm.group(2) title = ( (tag.get("aria-label") or "").strip() or (tag.get_text() or "").strip() or f"Simbli meeting {mid}" ) rel = f"/SB_Meetings/ViewMeeting.aspx?S={sid}&MID={mid}" push(urljoin(base_root, rel), title) mm = _VIEWMINUTES_ONCLICK_RE.search(oc) if mm: sid, mid, status = mm.group(1), mm.group(2), (mm.group(3) or "").strip().upper() if status != "PUBLISHED": continue title = ( (tag.get("aria-label") or "").strip() or (tag.get_text() or "").strip() or f"Simbli published minutes {mid}" ) rel = f"/SB_Meetings/ViewMeeting.aspx?S={sid}&MID={mid}&T=1" push(urljoin(base_root, rel), title) return out def extract_simbli_agenda_minutes_html_pairs( html: str, page_url: str, homepage: str, ) -> List[Tuple[str, str]]: """ Extract absolute HTTP(S) links to Simbli HTML agenda/minutes pages for PDF rendering. Includes **the ViewMeeting hub URL itself** when ``page_url`` is ``ViewMeeting.aspx?MID=…`` — agendas and minutes usually render inside that shell; nested ```` alone often misses everything. Same-origin / trusted-host rules mirror PDF extraction: homepage, current page, or trusted offsite. """ soup = _beautifulsoup_meetings(html, page_url=page_url, log_label="extract_simbli_html") if soup is None: return [] join_base = document_join_base(page_url, soup) out: List[Tuple[str, str]] = [] seen: Set[str] = set() try: pq_self = _path_query_lower(page_url) except Exception: pq_self = "" if ( is_simbli_eboard_host(page_url) and "viewmeeting.aspx" in pq_self and "mid=" in pq_self ): nu_self = urlunparse(urlparse(page_url)._replace(fragment="")) if nu_self not in seen: seen.add(nu_self) minutes_tab = bool(re.search(r"(?:^|[?&])t=1(?:&|$)", pq_self)) label = ( "Simbli ViewMeeting published minutes" if minutes_tab else "Simbli ViewMeeting packet" ) out.append((nu_self, label)) def eligible_target(full: str) -> bool: if not full.lower().startswith(("http://", "https://")): return False if not ( is_same_site(full, homepage) or is_same_site(full, page_url) or is_trusted_offsite(full) ): return False return True for a in soup.find_all("a", href=True): href = (a.get("href") or "").strip() if not href or href.startswith("#") or href.lower().startswith("javascript:"): continue full = resolve_page_href(join_base, href) text = _link_anchor_text(a) if not eligible_target(full): continue if not is_simbli_agenda_minutes_html_target(full, text): continue nu = urlunparse(urlparse(full)._replace(fragment="")) if nu in seen: continue seen.add(nu) out.append((nu, text)) # onclick / JS redirects occasionally expose agenda URLs without proper href — skip for safety. return out # Counties sometimes link a short branded host (e.g. ``tallaco.com/commission-meetings/``) that is # not the GSA/NACO ``website_url`` host. Only follow when the href is present in HTML and path # looks like a commission/board meeting archive — no blind host generation. _LINKED_MEETING_ARCHIVE_PATH_MARKERS: Tuple[str, ...] = ( "commission-meeting", "commission_meeting", "board-meeting", "board_meeting", "/meetings/", "/minutes/", "proceedings", "action-minutes", "action_minutes", "/agendas/", "commission-minute", "meeting-archive", "meeting_archive", "county-commission", "county_commission", ) _THIRD_PARTY_MEETING_NAV_EXCLUDED_HOST_SUFFIXES: Tuple[str, ...] = ( "facebook.com", "fb.com", "twitter.com", "x.com", "instagram.com", "linkedin.com", "youtube.com", "youtu.be", "google.com", "tiktok.com", "pinterest.com", "yelp.com", "wikipedia.org", "amazon.com", ) def is_linked_local_meeting_microsite(url: str, homepage: str) -> bool: """ True for http(s) URLs on a host **different** from ``homepage`` whose path/query suggests a local meeting archive (linked from the county site). Excludes known social / search hosts and vendor stacks handled by :func:`is_vendor_meeting_page_url`. """ if not url or not homepage: return False if is_same_site(url, homepage): return False if is_trusted_offsite(url): return False if is_amazonaws_document_cdn_host(url): # S3 / CloudFront-style hosts: meeting materials are downloaded via ``MEETING_DOWNLOAD_EXT``, # not crawled as HTML (paths like ``/Minutes/…`` would otherwise match archive markers). return False try: p = urlparse(url) if p.scheme not in ("http", "https"): return False host = p.netloc.lower().split(":")[0].rstrip(".") if not host or "." not in host: return False if any( host == s or host.endswith("." + s) for s in _THIRD_PARTY_MEETING_NAV_EXCLUDED_HOST_SUFFIXES ): return False pq = _path_query_lower(url) return any(m in pq for m in _LINKED_MEETING_ARCHIVE_PATH_MARKERS) except Exception: return False PDF_EXT = re.compile(r"\.pdf(\?|#|$)", re.I) # Meeting materials linked from government pages (same-site, vendor, or :func:`is_amazonaws_document_cdn_host`). # ``.doc`` is matched with a trailing boundary so ``.docker`` / ``.docs`` style paths are not treated as Word. MEETING_DOWNLOAD_EXT = re.compile( r"\.(?:pdf|docx|rtf|pptx?|mp3|m4a|wav)(?:\?|#|$)|\.doc(?:\?|#|$)", re.I, ) # Revize / classic ASP / similar CMS “site search” portals (e.g. Autauga # ``Default.asp?ID=122&pg=Site+Search&action=search``). _SITE_SEARCH_URL_RE = re.compile( r"(?:^|[?&])pg=site\+search" r"|(?:^|[?&])pg=site%20search" r"|action=search" r"|site\+search" r"|site%20search" r"|/site[_\-]?search" r"|sitesearch\.asp" r"|pagesearch", re.I, ) def strip_url_fragment(url: str) -> str: p = urlparse(url) return urlunparse((p.scheme, p.netloc, p.path, p.params, p.query, "")) def dedupe_repeated_url_path(url: str) -> str: """ Collapse duplicated consecutive path blocks, e.g. ``/departments/board/departments/board/index.php`` -> ``/departments/board/index.php``. Some CMS pages use site-root-style relative hrefs (``departments/...``) **without** a leading ``/``; :func:`urllib.parse.urljoin` then resolves them under the current directory and repeats the path prefix (Burke County GA ``board_of_commissioners`` nav, etc.). """ if not url or not url.strip(): return url try: p = urlparse(url) path = p.path or "/" parts = [s for s in path.split("/") if s != ""] min_block = 2 if len(parts) < min_block * 2: return url changed = True while changed and len(parts) >= min_block * 2: changed = False max_block = len(parts) // 2 for block in range(max_block, min_block - 1, -1): if parts[:block] == parts[block : 2 * block]: parts = parts[block:] changed = True break new_path = "/" + "/".join(parts) if parts else "/" if new_path == path: return url return urlunparse((p.scheme, p.netloc, new_path, p.params, p.query, "")) except Exception: return url def resolve_page_href(page_url: str, href: str) -> str: """Absolute URL from ``page_url`` + ``href``, strip fragment, fix duplicated path segments.""" full = strip_url_fragment(urljoin((page_url or "").strip(), (href or "").strip())) return dedupe_repeated_url_path(full) def document_join_base(page_url: str, soup: BeautifulSoup) -> str: """ Base URL for resolving relative ``href``/``src`` in this document. When the page has ```` (common on Revize / IIS sites), relative links like ``Documents/…`` must join to that base — not to ``…/Default.asp?…``, or ``urljoin`` drops the site subdirectory (e.g. Autauga County ``/Sites/Autauga_County/``). """ pu = (page_url or "").strip() tag = soup.find("base", href=True) if tag: b = (tag.get("href") or "").strip() if b and not b.lower().startswith("javascript:"): return resolve_page_href(pu, b) return pu def site_search_portal_variants(portal_url: str) -> List[str]: """ For same-site search portal URLs, return the base URL plus a few GET variants with meeting keywords (CMS-specific; harmless 404s are dropped later). """ base = strip_url_fragment((portal_url or "").strip()) if not base: return [] low = base.lower() if "content-search" in low and ("keyword" in low or "dlv_" in low): return opencivic_content_search_query_variants(base) if not _SITE_SEARCH_URL_RE.search(base): return [base] sep = "&" if "?" in base else "?" out: List[str] = [base] for param in ( "keywords=meetings", "keyword=meetings", "search=meetings", "q=meetings", "Search=meetings", "txtSearch=meetings", "criteria=meetings", "SearchString=meetings", "keywords=youtube", "keywords=video", "keywords=webcast", "search=live+stream", ): cand = f"{base}{sep}{param}" if cand not in out: out.append(cand) return out def html_suggests_wordpress_site(html: str) -> bool: """ Heuristic: HTML likely comes from WordPress, so root ``/?s=`` site search is plausible. Avoids hammering non-WP sites (e.g. Revize, Granicus OpenCivic) with WordPress-only URLs. """ if not html: return False sample = html[:600_000] low = sample.lower() if "/wp-json/" in low or "/wp-content/" in low or "/wp-includes/" in low: return True if "wp-embed.min.js" in low or 'content="wordpress' in low: return True if re.search(r']+rel=["\']https://api\.w\.org/', sample, re.I): return True if re.search( r']{0,600}action=["\'][^"\']*["\'][^>]{0,1200}name=["\']s["\']', sample, re.I | re.DOTALL, ): return True if re.search( r'name=["\']s["\'][^>]{0,400}action=["\'][^"\']*[\?&]s=', sample, re.I | re.DOTALL, ): return True return False def extract_opencivic_content_search_portals(html: str, page_url: str, homepage: str) -> List[str]: """ Granicus / OpenCivic-style site search, e.g. ``/Content-search?dlv_…Public%20Site%20Search=(keyword=)`` (filled per-query later). Example host: ``dallascounty-al.org`` (Granicus OpenCivic ``Content-search``). """ soup = _beautifulsoup_meetings(html, page_url=page_url, log_label="opencivic_content_search") if soup is None: return [] join_base = document_join_base(page_url, soup) found: List[str] = [] seen: Set[str] = set() def consider(raw: str) -> None: u0 = (raw or "").strip() if not u0 or u0.startswith("#") or u0.lower().startswith("javascript:"): return if "content-search" not in u0.lower(): return if "keyword" not in u0.lower() and "dlv_" not in u0.lower(): return full = resolve_page_href(join_base, u0) if not is_same_site(full, homepage): return if full not in seen: seen.add(full) found.append(full) for a in soup.find_all("a", href=True): consider(a.get("href") or "") for form in soup.find_all("form", action=True): if (form.get("method") or "get").upper() != "GET": continue consider(form.get("action") or "") return found def opencivic_content_search_query_variants( template_url: str, keywords: Optional[Sequence[str]] = None, ) -> List[str]: """ Turn one ``Content-search`` template into concrete GET URLs by filling ``(keyword=…)``. If there is no ``(keyword=)`` placeholder, returns the template unchanged (single URL). """ base = strip_url_fragment((template_url or "").strip()) if not base: return [] terms: Tuple[str, ...] = tuple(keywords) if keywords else ( "meetings", "minutes", "agenda", "board", "council", "commission", "youtube", "video", "webcast", "live", "meeting minutes", ) if "(keyword=" not in base and "(keyword%3d=" not in base.lower(): return [base] out: List[str] = [] for kw in terms: filled = re.sub( r"\(keyword=\)", f"(keyword={quote_plus(kw)})", base, count=1, flags=re.I, ) if filled != base: out.append(filled) return list(dict.fromkeys(out)) def extract_site_search_portal_urls(html: str, page_url: str, homepage: str) -> List[str]: """ Discover same-site links to vendor “site search” pages (ASP ``Default.asp``, etc.). These are not covered by WordPress ``/?s=`` heuristics; we scrape the homepage/nav HTML for ``pg=Site+Search``, ``action=search``, and similar patterns. """ soup = _beautifulsoup_meetings(html, page_url=page_url, log_label="site_search_portal") if soup is None: return [] found: List[str] = [] seen: Set[str] = set() join_base = document_join_base(page_url, soup) def add(raw: str) -> None: u = resolve_page_href(join_base, raw.strip()) if not u or u in seen: return if not is_same_site(u, homepage): return if not _SITE_SEARCH_URL_RE.search(u): return seen.add(u) found.append(u) for a in soup.find_all("a", href=True): href = (a.get("href") or "").strip() if not href or href.startswith("#") or href.lower().startswith("javascript:"): continue full = resolve_page_href(join_base, href) if _SITE_SEARCH_URL_RE.search(full) or _SITE_SEARCH_URL_RE.search(href): add(full) continue text = (a.get_text() or "").strip().lower() if "site search" in text and "search" in full.lower(): add(full) for form in soup.find_all("form", action=True): action = (form.get("action") or "").strip() if not action: continue if (form.get("method") or "get").upper() != "GET": continue full = resolve_page_href(join_base, action) if is_same_site(full, homepage) and _SITE_SEARCH_URL_RE.search(full): add(full) for u in extract_opencivic_content_search_portals(html, page_url, homepage): if u not in seen: seen.add(u) found.append(u) return found def _is_youtube_related_host(host: str) -> bool: h = (host or "").lower().split(":")[0].lstrip(".") if h in ("youtu.be", "youtube.com", "www.youtube.com", "m.youtube.com", "music.youtube.com"): return True if h.endswith(".youtube.com") or h.endswith(".youtu.be"): return True if "youtube-nocookie.com" in h: return True return False def normalize_youtube_url(raw: str, page_url: str) -> Optional[str]: """Return canonical ``https`` YouTube URL without fragment, or ``None`` if not YouTube.""" u = (raw or "").strip() if not u or u.lower().startswith(("javascript:", "mailto:", "tel:")): return None if u.startswith("//"): u = "https:" + u u = dedupe_repeated_url_path(strip_url_fragment(urljoin(page_url, u))) try: p = urlparse(u) except Exception: return None host = (p.netloc or "").lower().split(":")[0] if not _is_youtube_related_host(host): return None scheme = "https" if host == "youtu.be": vid = (p.path or "").strip("/").split("/")[0] if not vid: return None vid = vid.split("?")[0] return strip_url_fragment(f"{scheme}://www.youtube.com/watch?v={vid}") netloc = host if "youtube-nocookie.com" in netloc: netloc = "www.youtube-nocookie.com" elif netloc in ("youtube.com", "m.youtube.com"): netloc = "www.youtube.com" return strip_url_fragment(urlunparse((scheme, netloc, p.path, p.params, p.query, ""))) def classify_youtube_link(url: str) -> str: """Rough bucket: ``video``, ``channel``, ``playlist``, or ``other``.""" low = (url or "").lower() if "/playlist" in low and "list=" in low: return "playlist" if "/playlist/" in low: return "playlist" if "/channel/" in low: return "channel" if "/@" in low: return "channel" if "/user/" in low: return "channel" if "/c/" in low and "/channel/" not in low: return "channel" if "/watch" in low or "/embed/" in low or "/shorts/" in low or "/live" in low: return "video" if "youtu.be/" in low: return "video" return "other" _YOUTUBE_MEETING_STRONG = re.compile( r"\b(?:city|town|village|borough)\s+council\b|" r"\bcounty\s+commission\b|\bboard\s+of\s+(?:commissioners|supervisors|trustees)\b|" r"\b(?:school|planning|zoning|park|library)\s+board\b|" r"\b(?:public|town)\s+hearing\b|" r"\b(?:committee|commission)\s+of\s+the\s+whole\b|" r"\b(?:regular|special|study|work)\s+session\b|\bworkshop\b|" r"\b(?:full|complete)\s+meeting\b|\bmeeting\s+(?:video|recording|replay)\b|" r"\b(?:live|recorded)\s+(?:stream|meeting)\b|" r"\bagenda\b|\bminutes\b|\bproceedings\b|\baction\s*minutes\b|\bwebcast\b|\bgovernment\s+meeting\b|" r"\bmunicipal\b|\b(?:city|county)\s+hall\b", re.I, ) _YOUTUBE_MEETING_WEAK = re.compile( r"\b(?:meetings?|hearing|session|council|commission|board|supervisor|trustee|" r"alderm(?:an|en)|mayor|clerk|agenda|minutes?|proceedings|action\s*minutes|townhall)\b", re.I, ) _YOUTUBE_NON_MEETING = re.compile( r"\b(?:music\s+video|official\s+video|lyrics|gameplay|minecraft|fortnite|" r"tutorial|asmr|unboxing|reaction\s+video|highlights\s+only|sports\s+highlights)\b", re.I, ) def youtube_url_for_oembed(url: str) -> Optional[str]: """ Normalize to a ``https://www.youtube.com/watch?v=…`` URL for YouTube's public oEmbed API. Returns ``None`` for channels, playlists, and other shapes oEmbed does not resolve to a single video. """ u = (url or "").strip() if not u: return None if classify_youtube_link(u) != "video": return None try: p = urlparse(u) except Exception: return None host = (p.netloc or "").lower().split(":")[0] path = p.path or "" q = p.query or "" if host in ("youtu.be", "www.youtu.be"): vid = path.strip("/").split("/")[0] return f"https://www.youtube.com/watch?v={vid}" if vid else None if "/embed/" in path: m = re.search(r"/embed/([^/?&]+)", path) if m: return f"https://www.youtube.com/watch?v={m.group(1)}" if "/shorts/" in path: m = re.search(r"/shorts/([^/?&]+)", path) if m: return f"https://www.youtube.com/watch?v={m.group(1)}" if "/watch" in path or path.rstrip("/").endswith("watch"): vids = parse_qs(q).get("v") or [] if vids and vids[0]: return strip_url_fragment(f"https://www.youtube.com/watch?v={vids[0]}") if "/live" in path: vids = parse_qs(q).get("v") or [] if vids and vids[0]: return strip_url_fragment(f"https://www.youtube.com/watch?v={vids[0]}") return None def score_youtube_meeting_relevance( title: str, author_name: str = "", *, homepage_host: str = "", ) -> Tuple[str, float, List[str]]: """ Score oEmbed ``title`` / ``author_name`` for civic-meeting similarity. Returns ``(label, score, signal_tags)`` where ``label`` is one of: ``likely_meeting``, ``possible_meeting``, ``weak_signal``, ``unlikely_meeting``, ``unclear``, or ``unknown`` (empty text). """ t = (title or "").strip() a = (author_name or "").strip() blob = f"{t}\n{a}".strip() signals: List[str] = [] if not blob: return "unknown", 0.0, [] score = 0.0 if _YOUTUBE_NON_MEETING.search(blob): score -= 2.5 signals.append("non_meeting_topic_hint") if _YOUTUBE_MEETING_STRONG.search(blob): score += 4.0 signals.append("strong_meeting_term") if _YOUTUBE_MEETING_WEAK.search(blob): score += 1.5 signals.append("weak_meeting_term") h = (homepage_host or "").lower().split(":")[0].lstrip(".") if h and h in a.lower(): score += 1.25 signals.append("author_matches_homepage_host") if score >= 3.5: label = "likely_meeting" elif score >= 2.0: label = "possible_meeting" elif score > 0: label = "weak_signal" elif score < 0: label = "unlikely_meeting" else: label = "unclear" return label, score, signals def extract_youtube_refs(html: str, page_url: str) -> List[Dict[str, str]]: """ Collect YouTube video and channel URLs from ````, ``