Update utils.py

utils.py

@@ -196,42 +196,105 @@ def _ts_normalize(t_val) -> Optional[float]:
         return None
 
 
-async def get_all_epg(auth: dict, channels=None, date=None, force=False):
-    if channels is None:
-        channels = await get_channels(auth)
-    jst = timezone(timedelta(hours=9))
-    if date is None:
-        target_date = datetime.now(jst).strftime('%Y-%m-%d')
-    else:
-        if len(str(date)) == 8 and str(date).isdigit():
-            target_date = f"{str(date)[:4]}-{str(date)[4:6]}-{str(date)[6:8]}"
-        else:
-            target_date = date
+async def get_all_epg(auth: Dict[str, Any], channels: list = None, date: str = None, force: bool = False) -> Dict[str, Any]:
+    cache_key_date = date if date else 'full'
     if not force:
-        cached = cache.get_epg('_all_', 'full')
+        cached = cache.get_epg('_all_', cache_key_date)
+        if cached:
+            return cached
+
+    try:
+        url = Config.get_list_url(auth['vms_uid'], with_epg=True)
+
+        headers = {
+            'Referer': Config.REQUIRED_REFERER,
+            'User-Agent': 'Mozilla/5.0'
+        }
+
+        async with httpx.AsyncClient(timeout=Config.TIMEOUT) as client:
+            response = await client.get(url, headers=headers)
+            response.raise_for_status()
+            data = response.json()
+
+            result = {}
+            jst = timezone(timedelta(hours=9))
+
+            if date:
+                try:
+                    target_dt = datetime.strptime(date, "%Y%m%d").replace(tzinfo=jst)
+                except ValueError:
+                    target_dt = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=jst)
+                start_ts = target_dt.timestamp()
+                end_ts = (target_dt + timedelta(days=1)).timestamp()
+            else:
+                start_ts = None
+                end_ts = None
+
+            source_channels = data.get('result', [])
+
+            for channel in source_channels:
+                channel_id = channel.get('id')
+                record_epg = channel.get('record_epg')
+
+                if not channel_id:
+                    continue
+
+                if not record_epg:
+                    result[channel_id] = {'data': []}
+                    continue
+
+                try:
+                    epg_list = json.loads(record_epg)
+
+                    processed_programs = []
+                    for i, program in enumerate(epg_list):
+                        if not program.get('time'):
+                            continue
+
+                        if 'time_end' not in program or not program['time_end']:
+                            if i + 1 < len(epg_list) and epg_list[i + 1].get('time'):
+                                program['time_end'] = epg_list[i + 1]['time']
+                            else:
+                                continue
+
+                        processed_programs.append(program)
+
+                    daily_epg = {}
+                    for program in processed_programs:
+                        ts = _ts_normalize(program['time'])
+                        if ts is None:
+                            continue
+                        dt = datetime.fromtimestamp(ts, tz=jst)
+                        date_str = dt.strftime('%Y-%m-%d')
+                        if date_str not in daily_epg:
+                            daily_epg[date_str] = []
+                        daily_epg[date_str].append(program)
+
+                    for d, programs in daily_epg.items():
+                        sorted_programs = sorted(programs, key=lambda x: x['time'])
+                        cache.set_epg(channel_id, d, sorted_programs)
+
+                    if start_ts is not None and end_ts is not None:
+                        filtered = [
+                            p for p in processed_programs
+                            if start_ts <= (_ts_normalize(p.get('time')) or 0) < end_ts
+                        ]
+                        result[channel_id] = {'data': sorted(filtered, key=lambda x: x['time'])}
+                    else:
+                        result[channel_id] = {'data': processed_programs}
+
+                except json.JSONDecodeError:
+                    result[channel_id] = {'data': []}
+                    continue
+
+            cache.set_epg('_all_', cache_key_date, result)
+            return result
+
+    except Exception as e:
+        cached = cache.get_epg('_all_', cache_key_date)
         if cached:
             return cached
-    sem = asyncio.Semaphore(10)
-    async def _fetch_one(ch):
-        ch_id = str(ch.get('id', ''))
-        if not ch_id:
-            return ch_id, []
-        if not force:
-            cached_ch = cache.get_epg(ch_id, target_date)
-            if cached_ch is not None:
-                return ch_id, cached_ch
-        async with sem:
-            try:
-                data = await fetch_epg(ch_id, target_date, auth)
-                if data:
-                    cache.set_epg(ch_id, target_date, data)
-                return ch_id, data or []
-            except Exception:
-                return ch_id, []
-    results = await asyncio.gather(*[_fetch_one(ch) for ch in channels])
-    all_epg = {ch_id: {"data": data} for ch_id, data in results if ch_id}
-    cache.set_epg('_all_', 'full', all_epg)
-    return all_epg
+        return {}
 
 
 def get_jst_date(dt: Optional[datetime] = None) -> str: