Upload auto_scheduler.py

auto_scheduler.py

@@ -27,25 +27,20 @@ SCHEDULE_STATE_FILE = os.path.join(DATA_DIR, 'scheduler_state.json')
 def _load_state():
     try:
         if os.path.exists(SCHEDULE_STATE_FILE):
-            with open(SCHEDULE_STATE_FILE, 'r') as f:
-                return json.load(f)
-    except Exception:
-        pass
+            with open(SCHEDULE_STATE_FILE, 'r') as f: return json.load(f)
+    except: pass
     return {}
 
 def _save_state(state):
     try:
         tmp = SCHEDULE_STATE_FILE + '.tmp'
-        with open(tmp, 'w') as f:
-            json.dump(state, f, ensure_ascii=False)
+        with open(tmp, 'w') as f: json.dump(state, f, ensure_ascii=False)
         os.replace(tmp, SCHEDULE_STATE_FILE)
-    except Exception as e:
-        LOG.warning(f"Cannot save scheduler state: {e}")
+    except Exception as e: LOG.warning(f"Cannot save state: {e}")
 
 _STOP = set('và của các những một được trong với cho tại sau trước khi không người việt nam hôm nay mới nhất nóng tin tức cập nhật theo từ đến là có thì này đã để'.split())
 
-def _clean(s):
-    return re.sub(r"\s+", " ", str(s or "")).strip()
+def _clean(s): return re.sub(r"\s+", " ", str(s or "")).strip()
 
 def _get_hot_topics():
     freq = {}; display = {}
@@ -79,8 +74,7 @@ def _get_hot_topics():
                             key = phrase.lower()
                             freq[key] = freq.get(key, 0) + 1
                             display[key] = phrase
-        except Exception:
-            continue
+        except: continue
     ranked = sorted(freq.items(), key=lambda x: x[1], reverse=True)
     topics = []; seen = set()
     for key, count in ranked:
@@ -96,21 +90,14 @@ def _get_hot_topics():
             topics.append({'label': '#' + re.sub(r'\s+', '', kw.title()), 'topic': kw, 'count': 0})
     return topics[:24]
 
-_ai_ext = None
-_ai_patch = None
-
+_ai_ext = None; _ai_patch = None
 def _get_ai_ext():
     global _ai_ext
-    if _ai_ext is None:
-        import ai_ext as m
-        _ai_ext = m
+    if _ai_ext is None: import ai_ext as m; _ai_ext = m
     return _ai_ext
-
 def _get_ai_patch():
     global _ai_patch
-    if _ai_patch is None:
-        import ai_patch as m
-        _ai_patch = m
+    if _ai_patch is None: import ai_patch as m; _ai_patch = m
     return _ai_patch
 
 _RSS_FEEDS = [
@@ -127,8 +114,7 @@ _RSS_FEEDS = [
 ]
 
 def _search_articles_by_topic(topic, limit=4):
-    all_articles = []
-    seen_urls = set()
+    all_articles = []; seen_urls = set()
     topic_lower = topic.lower()
     topic_words = set(re.findall(r'[A-Za-zÀ-ỹ0-9]+', topic_lower))
     for feed_url, source in _RSS_FEEDS:
@@ -157,18 +143,13 @@ def _search_articles_by_topic(topic, limit=4):
                             ogi = art_soup.find('meta', property='og:image')
                             if ogi: img = ogi.get('content', '')
                         except: pass
-                    all_articles.append({
-                        'title': title, 'url': link, 'raw': desc or title, 'image': img,
-                        'via': source,
-                        'source': {'title': title, 'url': link, 'excerpt': (desc or title)[:700], 'via': source}
-                    })
+                    all_articles.append({'title': title, 'url': link, 'raw': desc or title, 'image': img, 'via': source, 'source': {'title': title, 'url': link, 'excerpt': (desc or title)[:700], 'via': source}})
                     if len(all_articles) >= limit: break
         except: continue
     return all_articles[:limit]
 
 async def _create_ai_post(topic):
-    ai_ext = _get_ai_ext()
-    ai_patch = _get_ai_patch()
+    ai_ext = _get_ai_ext(); ai_patch = _get_ai_patch()
     articles = _search_articles_by_topic(topic, limit=4)
     if not articles:
         LOG.warning(f"No articles for topic: {topic}. Fallback.")
@@ -200,9 +181,8 @@ async def _create_ai_post(topic):
             except: pass
             posts.append(post)
         except Exception as e:
-            LOG.error(f"Error creating post: {e}")
-    if not posts:
-        return await _create_fallback_post(topic, ai_ext, ai_patch)
+            LOG.error(f"Error post: {e}")
+    if not posts: return await _create_fallback_post(topic, ai_ext, ai_patch)
     wall = posts + wall
     ai_ext._save_ai_wall(wall)
     for post in posts:
@@ -211,7 +191,7 @@ async def _create_ai_post(topic):
     return posts
 
 async def _create_fallback_post(topic, ai_ext, ai_patch):
-    LOG.info(f"Fallback post: {topic}")
+    LOG.info(f"Fallback: {topic}")
     try:
         text = f"• {topic} đang là chủ đề nóng hôm nay.\n• Theo dõi VNEWS để cập nhật tin tức mới nhất."
         img = ai_ext.pollination_image_url(topic)
@@ -244,8 +224,7 @@ def _try_generate_short(post):
                 if p.get('id') == post_id: wall[i] = post; break
             ai_ext._save_ai_wall(wall); return
         threading.Thread(target=lambda: _generate_short_worker(post, segments, post_id, suffix, out_mp4), daemon=True).start()
-    except Exception as e:
-        LOG.warning(f"Short init: {e}")
+    except Exception as e: LOG.warning(f"Short init: {e}")
 
 def _generate_short_worker(post, segments, post_id, suffix, out_mp4):
     import subprocess
@@ -292,10 +271,8 @@ def _generate_short_worker(post, segments, post_id, suffix, out_mp4):
                 if p.get('id') == post_id: wall[i] = post; break
             ai_ext._save_ai_wall(wall)
             LOG.info(f"Short: {post_id}")
-    except Exception as e:
-        LOG.warning(f"Short fail: {e}")
+    except Exception as e: LOG.warning(f"Short fail: {e}")
 
-# ===== FIX v4: _run_async helper - works from both sync and async contexts =====
 def _run_async(coro):
     """Run async coroutine safely regardless of current event loop state."""
     try:
@@ -308,10 +285,90 @@ def _run_async(coro):
 
 def _run_scheduled_posting():
     LOG.info("=" * 50)
-    LOG.info("Auto scheduler triggered at %s", datetime.now(VN_TZ).strftime('%H:%M %d/%m/%Y'))
+    LOG.info("Scheduler triggered at %s", datetime.now(VN_TZ).strftime('%H:%M %d/%m/%Y'))
     LOG.info("=" * 50)
     try:
         hot_topics = _get_hot_topics()
-        if not hot_topics: return
-        selected = []
-        seen_labels = set()
+        if not hot_topics:
+            LOG.warning("No hot topics"); return
+        selected = []; seen_labels = set()
+        for t in hot_topics:
+            label = t.get('label', '')
+            if label and label not in seen_labels:
+                seen_labels.add(label); selected.append(t['topic'])
+            if len(selected) >= 3: break
+        if len(selected) < 3:
+            selected = ['Thời sự Việt Nam', 'Kinh tế Việt Nam', 'Thể thao']
+        LOG.info(f"Topics: {selected}")
+        async def _do_all():
+            results = []
+            for topic in selected:
+                try:
+                    posts = await _create_ai_post(topic)
+                    results.append({'topic': topic, 'posts': len(posts) if posts else 0})
+                    LOG.info(f"{'✓' if posts else '✗'} {topic}: {len(posts) if posts else 0} posts")
+                except Exception as e:
+                    LOG.error(f"Error {topic}: {e}")
+                    results.append({'topic': topic, 'posts': 0})
+            return results
+        results = _run_async(_do_all())
+        LOG.info(f"Done: {len(results)} topics")
+        for r in results: LOG.info(f"  • {r['topic']}: {r['posts']} bài")
+    except Exception as e:
+        LOG.error(f"Scheduler error: {e}", exc_info=True)
+
+def _check_missed_slots():
+    try:
+        state = _load_state()
+        today_str = datetime.now(VN_TZ).strftime('%Y-%m-%d')
+        now = datetime.now(VN_TZ); cur_mins = now.hour * 60 + now.minute
+        ran = state.get(today_str, {})
+        for s in SCHEDULE_TIMES:
+            lbl = SCHEDULE_LABELS[s]; sm = s[0] * 60 + s[1]
+            if ran.get(lbl): continue
+            if cur_mins >= sm:
+                LOG.info(f"Catch-up: {lbl}")
+                _run_scheduled_posting()
+                if today_str not in state: state[today_str] = {}
+                state[today_str][lbl] = True; _save_state(state)
+    except Exception as e: LOG.error(f"Catch-up: {e}")
+
+def _scheduler_loop():
+    LOG.info("Scheduler started")
+    LOG.info(f"Schedule: {', '.join(f'{h:02d}:{m:02d}' for h,m in SCHEDULE_TIMES)} VN")
+    state = _load_state(); today_str = datetime.now(VN_TZ).strftime('%Y-%m-%d')
+    ran = state.get(today_str, {})
+    now = datetime.now(VN_TZ); cur_mins = now.hour * 60 + now.minute
+    for s in SCHEDULE_TIMES:
+        lbl = SCHEDULE_LABELS[s]; sm = s[0] * 60 + s[1]
+        if ran.get(lbl): LOG.info(f"  ✓ {lbl} done"); continue
+        if cur_mins >= sm:
+            LOG.info(f"  → {lbl} missed! Catch-up")
+            _run_scheduled_posting()
+            if today_str not in state: state[today_str] = {}
+            state[today_str][lbl] = True; _save_state(state)
+        else: LOG.info(f"  ⏩ {lbl} upcoming")
+    while True:
+        try:
+            now = datetime.now(VN_TZ)
+            ck = (now.hour, now.minute)
+            state = _load_state(); today_str = now.strftime('%Y-%m-%d')
+            ran = state.get(today_str, {})
+            for s in SCHEDULE_TIMES:
+                lbl = SCHEDULE_LABELS[s]
+                if ck == s and not ran.get(lbl):
+                    LOG.info(f"On-time: {lbl}")
+                    _run_scheduled_posting()
+                    if today_str not in state: state[today_str] = {}
+                    state[today_str][lbl] = True; _save_state(state)
+                break
+            time.sleep(60)
+        except Exception as e:
+            LOG.error(f"Loop: {e}")
+            time.sleep(60)
+
+def start_auto_scheduler():
+    t = threading.Thread(target=_scheduler_loop, daemon=True, name="auto-scheduler")
+    t.start()
+    LOG.info("Auto scheduler started")
+    return t