Spaces:
Running
Running
| """VNEWS — Bongda Proxy Endpoint (for fast match detail loading)""" | |
| import requests | |
| from bs4 import BeautifulSoup | |
| import re | |
| import json | |
| def _cl(s): | |
| return re.sub(r'\s+', ' ', str(s or '')).strip() | |
| def _normalize_time(raw): | |
| t = _cl(raw) | |
| t = re.sub(r"(\d+)'\s*\+(\d+)", r"\1+\2'", t) | |
| t = t.replace("''", "'") | |
| return t | |
| def scrape_match_html(event_id, url=None): | |
| result = {"event_id": event_id, "found": False, "sections": []} | |
| headers = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", | |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", | |
| "Referer": "https://bongda.com.vn/", | |
| } | |
| html = None | |
| urls_to_try = [url] if url else [] | |
| urls_to_try += [ | |
| f"https://bongda.com.vn/tran-dau/{event_id}/centre/", | |
| f"https://bongda.com.vn/tran-dau/{event_id}/preview/", | |
| ] | |
| for u in urls_to_try: | |
| if not u: | |
| continue | |
| try: | |
| resp = requests.get(u, headers=headers, timeout=15, allow_redirects=True) | |
| if resp.status_code == 200 and len(resp.text) > 1000: | |
| html = resp.text | |
| break | |
| except Exception: | |
| continue | |
| if not html: | |
| return result | |
| try: | |
| soup = BeautifulSoup(html, 'html.parser') | |
| info = {} | |
| tel = soup.select_one('.teams') | |
| if tel: | |
| he = tel.select_one('.team.home') | |
| if he: | |
| ne = he.select_one('p:not(.logo)') or he.find('p') | |
| if ne: info['home_team'] = _cl(ne.get_text()) | |
| lo = he.select_one('img') | |
| if lo: info['home_logo'] = lo.get('src', '') | |
| ae = tel.select_one('.team.away') | |
| if ae: | |
| ne = ae.select_one('p:not(.logo)') or ae.find('p') | |
| if ne: info['away_team'] = _cl(ne.get_text()) | |
| lo = ae.select_one('img') | |
| if lo: info['away_logo'] = lo.get('src', '') | |
| sc = tel.select_one('.score') | |
| if sc: | |
| parts = [_cl(p.get_text()) for p in sc.select('p')] | |
| if len(parts) >= 2: info['score'] = f"{parts[0]} - {parts[1]}" | |
| lb = sc.select_one('.label') | |
| if lb: info['status_label'] = _cl(lb.get_text()) | |
| if info.get('home_team') and info.get('away_team'): | |
| result['info'] = info | |
| result['found'] = True | |
| result['sections'].append('info') | |
| else: | |
| return result | |
| events = [] | |
| events_div = soup.select_one('.events') | |
| if events_div: | |
| period = '' | |
| for child in events_div.children: | |
| if not hasattr(child, 'name') or not child.name: continue | |
| cls = ' '.join(child.get('class', [])) | |
| if 'period' in cls: | |
| h2 = child.find('h2') | |
| if h2: period = _cl(h2.get_text()) | |
| for ev in child.children: | |
| if not hasattr(ev, 'name') or not ev.name: continue | |
| ev_cls = ' '.join(ev.get('class', [])) | |
| if 'event' not in ev_cls: continue | |
| ev_data = {'team': 'home' if 'home' in ev_cls else 'away', 'period': period, 'type': 'unknown', 'time': ''} | |
| type_el = ev.select_one('.event-type') | |
| if type_el: | |
| if type_el.select_one('[class*="redcard"]'): ev_data['type'] = 'redcard' | |
| elif type_el.select_one('[class*="yellowcard"]'): ev_data['type'] = 'yellowcard' | |
| elif type_el.select_one('[class*="goal"]'): ev_data['type'] = 'goal' | |
| elif type_el.select_one('[class*="substitution"]'): ev_data['type'] = 'substitution' | |
| players_el = ev.select_one('.players') | |
| if players_el: | |
| time_el = players_el.select_one('.event-time') | |
| if time_el: ev_data['time'] = _normalize_time(time_el.get_text()) | |
| text = _cl(players_el.get_text(' ', strip=True).replace(ev_data['time'], '').strip()) | |
| ev_data['players'] = text | |
| events.append(ev_data) | |
| if events: | |
| result['events'] = events | |
| result['sections'].append('events') | |
| except Exception as e: | |
| result['error'] = str(e) | |
| return result | |
| from fastapi import Query | |
| from fastapi.responses import JSONResponse | |
| def add_bongda_proxy_endpoint(app): | |
| def proxy_bongda(event_id: int = Query(default=None), url: str = Query(default=None)): | |
| if event_id is None: | |
| return JSONResponse({'error': 'event_id required'}, status_code=400) | |
| return JSONResponse(scrape_match_html(event_id, url)) | |