# -*- coding: utf-8 -*- import threading import traceback from datetime import datetime import requests from flask import jsonify, render_template from plugin import PluginModuleBase from framework import F from .model import ModelFetchLog, ModelFreeGameItem, ModelSetting from . import scraper from .setup import P logger = P.logger package_name = P.package_name _fetch_lock = threading.Lock() SOURCE_LABELS = { "epic": "Epic", "steam": "Steam", "gog": "GOG", "indiegala": "IndieGala", "stove": "STOVE", "cheapshark": "CheapShark", } def _truthy(value): return str(value).lower() == "true" def _enabled_sources(): enabled = [] for source in SOURCE_LABELS: if _truthy(ModelSetting.get(f"source_{source}_enabled")): enabled.append(source) return enabled def _split_source_payload(results): grouped = {source: [] for source in SOURCE_LABELS} for source, items in (results or {}).items(): normalized_source = "indiegala" if source == "indiegala_free" else source if normalized_source not in grouped: continue for item in items or []: is_free = bool(item.get("is_free_period")) or float(item.get("current_price") or 0) == 0 if not is_free: continue if str(item.get("platform") or "") not in ["", normalized_source]: continue item["platform"] = normalized_source grouped[normalized_source].append(item) return grouped def _discord_send(webhook_url, games): lines = [] for game in games[:10]: prefix = "๐Ÿ†• NEW " if game.get("is_new") else "" title = game.get("title") or "Unknown" platform = SOURCE_LABELS.get(game.get("platform"), game.get("platform")) store_url = game.get("store_url") or "" score = int(game.get("metacritic_score") or 0) line = f"{prefix}**{title}** ({platform})" if score > 0: line += f" ยท MC {score}" if store_url: line += f"\n{store_url}" lines.append(line) content = "**๋ฌด๋ฃŒ ๊ฒŒ์ž„ ์•Œ๋ฆผ**\n\n" + "\n\n".join(lines) requests.post(webhook_url, json={"content": content}, timeout=10).raise_for_status() def _telegram_send(bot_token, chat_id, games): lines = [] for game in games[:10]: prefix = "๐Ÿ†• NEW " if game.get("is_new") else "" title = game.get("title") or "Unknown" platform = SOURCE_LABELS.get(game.get("platform"), game.get("platform")) store_url = game.get("store_url") or "" score = int(game.get("metacritic_score") or 0) line = f"{prefix}{title} ({platform})" if score > 0: line += f" ยท MC {score}" if store_url: line += f"\n{store_url}" lines.append(line) requests.post( f"https://api.telegram.org/bot{bot_token}/sendMessage", json={ "chat_id": chat_id, "text": "๋ฌด๋ฃŒ ๊ฒŒ์ž„ ์•Œ๋ฆผ\n\n" + "\n\n".join(lines), "parse_mode": "HTML", "disable_web_page_preview": True, }, timeout=10, ).raise_for_status() class Logic(PluginModuleBase): db_default = { "main_auto_start": "False", "main_interval": "0 */2 * * *", "auto_start": "False", "auto_interval": "0 */2 * * *", "notify_discord_webhook": "", "notify_telegram_bot_token": "", "notify_telegram_chat_id": "", "notify_enabled": "False", "source_epic_enabled": "True", "source_steam_enabled": "True", "source_gog_enabled": "True", "source_indiegala_enabled": "True", "source_stove_enabled": "True", "source_cheapshark_enabled": "True", "last_fetch_started": "", "last_fetch_finished": "", } def __init__(self, PM): super().__init__(PM, name="main", first_menu="setting", scheduler_desc="FreeGame fetch") def plugin_load(self): self._migrate_scheduler_settings() ModelFreeGameItem.ensure_schema() def _migrate_scheduler_settings(self): legacy_interval = str(ModelSetting.get("auto_interval") or "").strip() if legacy_interval and str(ModelSetting.get("main_interval") or "").strip() in ["", "0 */2 * * *"]: ModelSetting.set("main_interval", legacy_interval) if _truthy(ModelSetting.get("auto_start")) and not _truthy(ModelSetting.get("main_auto_start")): ModelSetting.set("main_auto_start", "True") def process_menu(self, sub, req): arg = ModelSetting.to_dict() arg["package_name"] = package_name arg["scheduler"] = str(F.scheduler.is_include(self.get_scheduler_name())) arg["is_running"] = str(F.scheduler.is_running(self.get_scheduler_name())) arg["source_labels"] = SOURCE_LABELS arg["platform_counts"] = ModelFreeGameItem.get_platform_counts() if sub == "list": return render_template("ff_freegame_main_list.html", arg=arg) if sub == "log": return render_template("log.html", package=package_name) return render_template("ff_freegame_main_setting.html", arg=arg) def process_ajax(self, sub, req): try: if sub == "setting_save": ret, _ = ModelSetting.setting_save(req) self.setting_save_after(None) ret["ret"] = "success" return jsonify(ret) if sub == "scheduler_toggle": if req.form["scheduler"] == "true": self.P.logic.scheduler_start(self.name) else: self.P.logic.scheduler_stop(self.name) return jsonify({"ret": "success"}) if sub == "execute_once": threading.Thread(target=self.scheduler_function, daemon=True).start() return jsonify({"ret": "success"}) if sub == "web_list": ModelFreeGameItem.ensure_schema() return jsonify(ModelFreeGameItem.web_list(req)) if sub == "platform_counts": return jsonify({"ret": "success", "data": ModelFreeGameItem.get_platform_counts()}) return jsonify({"ret": "error", "log": f"unsupported ajax: {sub}"}) except Exception as e: logger.error("Exception:%s", e) logger.error(traceback.format_exc()) return jsonify({"ret": "error", "log": str(e)}) def setting_save_after(self, change_list): if F.scheduler.is_include(self.get_scheduler_name()): self.P.logic.scheduler_stop(self.name) self.P.logic.scheduler_start(self.name) elif _truthy(ModelSetting.get("main_auto_start")): self.P.logic.scheduler_start(self.name) def scheduler_function(self): if not _fetch_lock.acquire(blocking=False): logger.info("FreeGame fetch skipped: already running") return try: with F.app.app_context(): logger.info("FreeGame scheduled fetch started") ModelSetting.set("last_fetch_started", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) ModelFreeGameItem.ensure_schema() enabled_sources = set(_enabled_sources()) results = scraper.fetch_all() grouped = _split_source_payload(results) fresh_free_games = [] for legacy_source in ["humble", "fanatical", "gmg", "directgames"]: ModelFreeGameItem.replace_source_items(legacy_source, []) for source, items in grouped.items(): if source not in enabled_sources: continue existing_ids = { row[0] for row in F.db.session.query(ModelFreeGameItem.external_id) .filter_by(platform=source) .all() } for item in items: item["is_new"] = str(item.get("external_id") or "") not in existing_ids ModelFreeGameItem.replace_source_items(source, items) ModelFetchLog(source, "ok", "", len(items)).save() logger.info("FreeGame source=%s saved=%d", source, len(items)) fresh_free_games.extend(items) disabled_sources = [source for source in SOURCE_LABELS if source not in enabled_sources] if disabled_sources: ModelFreeGameItem.delete_not_in_sources(enabled_sources) logger.info("FreeGame fetch completed: free_candidates=%d enabled_sources=%d", len(fresh_free_games), len(enabled_sources)) ModelFetchLog("all", "ok", f"free_candidates={len(fresh_free_games)} enabled_sources={len(enabled_sources)}", len(fresh_free_games)).save() ModelSetting.set("last_fetch_finished", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) self._notify(fresh_free_games) except Exception as e: logger.error("Exception:%s", e) logger.error(traceback.format_exc()) ModelFetchLog("all", "error", str(e), 0).save() finally: _fetch_lock.release() def _notify(self, games): if _truthy(ModelSetting.get("notify_enabled")) is False: return targets = list(games or []) if len(targets) == 0: return discord_webhook = ModelSetting.get("notify_discord_webhook") telegram_bot_token = ModelSetting.get("notify_telegram_bot_token") telegram_chat_id = ModelSetting.get("notify_telegram_chat_id") if discord_webhook: try: _discord_send(discord_webhook, targets) logger.info("FreeGame Discord notification sent: %d", min(len(targets), 10)) except Exception as e: logger.error("FreeGame Discord notification failed: %s", e) if telegram_bot_token and telegram_chat_id: try: _telegram_send(telegram_bot_token, telegram_chat_id, targets) logger.info("FreeGame Telegram notification sent: %d", min(len(targets), 10)) except Exception as e: logger.error("FreeGame Telegram notification failed: %s", e)