From 7bb9bfc97e54e8ad72a88e97988cc5aba2e1cfc5 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 1 Aug 2026 16:49:49 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20Sage-compliant=20DSPY=20=E2=80=94=20no?= =?UTF-8?q?=20imports,=20env=20delegation,=20init.py,=20scripts/load=5Fpat?= =?UTF-8?q?h.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add hotspot/init.py: load_hotspot() registers run_fetch/run_analysis/get_stats on ServerEnv - Rewrite .dspy files: only import json, use get_sor_context + env.*() delegation - Update db.py/engine.py/analysis.py: accept optional sor parameter for DSPY context - Add scripts/load_path.py: per-module RBAC permission registration - Follows sage-dspy-development + rbac-permission-initialization-pattern skills --- hotspot/__init__.py | 11 +++-- hotspot/analysis.py | 63 +++++++++++++++--------- hotspot/db.py | 89 +++++++++++++++++++++------------- hotspot/engine.py | 14 +++--- hotspot/init.py | 20 ++++++++ scripts/load_path.py | 70 ++++++++++++++++++++++++++ wwwroot/hotspot/analyze.dspy | 7 ++- wwwroot/hotspot/fetch_now.dspy | 7 ++- wwwroot/hotspot/stats.dspy | 7 ++- 9 files changed, 213 insertions(+), 75 deletions(-) create mode 100644 hotspot/init.py create mode 100644 scripts/load_path.py diff --git a/hotspot/__init__.py b/hotspot/__init__.py index 8a0dfa4..0a10223 100644 --- a/hotspot/__init__.py +++ b/hotspot/__init__.py @@ -1,14 +1,15 @@ """ 热点雷达 (Hotspot Radar) — Sage 框架模块 -用法: - from hotspot.engine import run_fetch - from hotspot.analysis import run_analysis - from hotspot.db import get_stats +集成方式: + # sage/app/sage.py + from hotspot.init import load_hotspot + # 在 init() 中添加: load_hotspot() """ +from .init import load_hotspot from .engine import run_fetch from .analysis import run_analysis from .db import get_stats from .parsers import BUILTIN as builtin_parsers -__all__ = ['run_fetch', 'run_analysis', 'get_stats', 'builtin_parsers'] +__all__ = ['load_hotspot', 'run_fetch', 'run_analysis', 'get_stats', 'builtin_parsers'] diff --git a/hotspot/analysis.py b/hotspot/analysis.py index aecae17..1c12f6d 100644 --- a/hotspot/analysis.py +++ b/hotspot/analysis.py @@ -94,13 +94,12 @@ async def analyze_item(item): return new_status, dims, detail -async def save_analysis(item_id, dims, detail, now=None): +async def save_analysis(item_id, dims, detail, now=None, sor=None): """保存/更新五维分析记录""" if now is None: now = now_str() - db = DBPools() - async with db.sqlorContext('sage') as sor: + async def _do(s): for dim, score in dims.items(): analysis_data = json.dumps({ 'dimension': dim, @@ -108,51 +107,71 @@ async def save_analysis(item_id, dims, detail, now=None): 'score': score, 'detail': detail, }, ensure_ascii=False) - - old = await sor.sqlExe( + old = await s.sqlExe( "select id from hotspot_analysis where item_id=${iid}$ and dimension=${dim}$", {'iid': item_id, 'dim': dim}) if old: - await sor.U('hotspot_analysis', { + await s.U('hotspot_analysis', { 'id': old[0]['id'], 'score': score, 'analysis_data': analysis_data, 'analyzed_at': now, }) else: - await sor.C('hotspot_analysis', { + await s.C('hotspot_analysis', { 'id': getID(), 'item_id': item_id, 'dimension': dim, 'score': score, 'analysis_data': analysis_data, 'analyzed_at': now, }) + if sor: + await _do(sor) + else: + db = DBPools() + async with db.sqlorContext('sage') as s: + await _do(s) -async def run_analysis(limit=500): - """批量分析最近的热点 (默认500条)""" - db = DBPools() + +async def run_analysis(limit=500, sor=None): + """批量分析最近的热点 (默认500条)。sor 可选，由 DSPY 提供。""" now = now_str() results = {'analyzed': 0, 'status_changes': 0} - async with db.sqlorContext('sage') as sor: - items = await sor.sqlExe( + async def _get_items(s): + return await s.sqlExe( "select * from hotspot_item order by first_seen desc limit ${n}$", {'n': limit}) + async def _update_item(s, item_id, adj_heat, velocity, new_status, now): + await s.U('hotspot_item', { + 'id': item_id, + 'heat_score': round(adj_heat, 2), + 'heat_velocity': velocity, + 'status': new_status, + 'last_updated': now, + }) + + if sor: + items = await _get_items(sor) + else: + db = DBPools() + async with db.sqlorContext('sage') as s: + items = await _get_items(s) + for item in items: current_status = item.get('status', 'emerging') new_status, dims, detail = await analyze_item(item) - - await save_analysis(item['id'], dims, detail, now) + await save_analysis(item['id'], dims, detail, now, sor=sor) if new_status != current_status: results['status_changes'] += 1 - async with db.sqlorContext('sage') as sor: - await sor.U('hotspot_item', { - 'id': item['id'], - 'heat_score': round(detail['adjusted_heat'], 2), - 'heat_velocity': detail['heat_velocity'], - 'status': new_status, - 'last_updated': now, - }) + adj_heat = detail['adjusted_heat'] + velocity = detail['heat_velocity'] + if sor: + await _update_item(sor, item['id'], adj_heat, velocity, new_status, now) + else: + db = DBPools() + async with db.sqlorContext('sage') as s: + await _update_item(s, item['id'], adj_heat, velocity, new_status, now) results['analyzed'] += 1 diff --git a/hotspot/db.py b/hotspot/db.py index 24fd635..2fec68b 100644 --- a/hotspot/db.py +++ b/hotspot/db.py @@ -1,5 +1,6 @@ """ 数据库操作: 保存热点条目、写抓取日志、统计查询 +所有函数接受可选 sor 参数 — 由 DSPY 的 get_sor_context 提供时不另开连接 """ from datetime import datetime from appPublic.uniqueID import getID @@ -8,32 +9,38 @@ from sqlor.dbpools import DBPools def now_str(): return datetime.now().strftime('%Y-%m-%d %H:%M:%S') -async def get_enabled_sources(): - """获取所有启用的来源""" +async def _get_sor(): + """standalone 模式：创建自己的数据库连接""" db = DBPools() - async with db.sqlorContext('sage') as sor: + return db.sqlorContext('sage') + +async def get_enabled_sources(sor=None): + """获取所有启用的来源""" + if sor: sources = await sor.R('hotspot_source', {}) + return [s for s in sources if s.get('enabled') == '1'] + async with await _get_sor() as s: + sources = await s.R('hotspot_source', {}) return [s for s in sources if s.get('enabled') == '1'] -async def save_items(items, source_id, fetch_time=None): +async def save_items(items, source_id, fetch_time=None, sor=None): """批量保存热点条目 (去重)""" if fetch_time is None: fetch_time = now_str() new_count = 0 dup_count = 0 - db = DBPools() - async with db.sqlorContext('sage') as sor: + async def _do(s): + nonlocal new_count, dup_count for item in items: url = item.get('url', '') if url: - existing = await sor.sqlExe( + existing = await s.sqlExe( "select id from hotspot_item where url=${u}$ limit 1", {'u': url}) if existing: dup_count += 1 continue - - await sor.C('hotspot_item', { + await s.C('hotspot_item', { 'id': getID(), 'source_id': source_id, 'title': item.get('title', '')[:500], 'url': (url or '')[:2000], @@ -53,46 +60,58 @@ async def save_items(items, source_id, fetch_time=None): }) new_count += 1 + if sor: + await _do(sor) + else: + async with await _get_sor() as s: + await _do(s) return new_count, dup_count -async def update_source_last_fetch(source_id): - db = DBPools() - async with db.sqlorContext('sage') as sor: +async def update_source_last_fetch(source_id, sor=None): + if sor: await sor.U('hotspot_source', {'id': source_id, 'last_fetch': now_str()}) + else: + async with await _get_sor() as s: + await s.U('hotspot_source', {'id': source_id, 'last_fetch': now_str()}) async def write_fetch_log(source_id, start_time, elapsed_ms, status, items_total, items_new, items_duplicate, - error_msg='', response_code=0, response_size_bytes=0): - db = DBPools() - async with db.sqlorContext('sage') as sor: - await sor.C('hotspot_fetch_log', { - 'id': getID(), - 'source_id': source_id, - 'start_time': start_time, - 'end_time': now_str(), - 'duration_ms': elapsed_ms, - 'status': status, - 'items_total': items_total, - 'items_new': items_new, - 'items_duplicate': items_duplicate, - 'error_msg': error_msg[:1000] if error_msg else '', - 'response_code': response_code, - 'response_size_bytes': response_size_bytes, - }) + error_msg='', response_code=0, response_size_bytes=0, + sor=None): + data = { + 'id': getID(), 'source_id': source_id, + 'start_time': start_time, 'end_time': now_str(), + 'duration_ms': elapsed_ms, 'status': status, + 'items_total': items_total, 'items_new': items_new, + 'items_duplicate': items_duplicate, + 'error_msg': error_msg[:1000] if error_msg else '', + 'response_code': response_code, 'response_size_bytes': response_size_bytes, + } + if sor: + await sor.C('hotspot_fetch_log', data) + else: + async with await _get_sor() as s: + await s.C('hotspot_fetch_log', data) -async def get_stats(): +async def get_stats(sor=None): """全维度统计""" from datetime import timedelta now = datetime.now() day_ago = (now - timedelta(hours=24)).strftime('%Y-%m-%d %H:%M:%S') - db = DBPools() - async with db.sqlorContext('sage') as sor: - items = await sor.R('hotspot_item', {'order': 'heat_score desc'}) - sources = await sor.R('hotspot_source', {}) - logs = await sor.sqlExe( + async def _do(s): + items = await s.R('hotspot_item', {'order': 'heat_score desc'}) + sources = await s.R('hotspot_source', {}) + logs = await s.sqlExe( "select * from hotspot_fetch_log where start_time >= ${t}$", {'t': day_ago}) + return items, sources, logs + + if sor: + items, sources, logs = await _do(sor) + else: + async with await _get_sor() as s: + items, sources, logs = await _do(s) status_count = {'emerging': 0, 'rising': 0, 'hot': 0, 'cooling': 0, 'expired': 0} for item in items: diff --git a/hotspot/engine.py b/hotspot/engine.py index 7fb18d1..2bae2b8 100644 --- a/hotspot/engine.py +++ b/hotspot/engine.py @@ -108,9 +108,9 @@ async def fetch_and_extract(session, src): return [], 0, 0 -async def run_fetch(): - """遍历所有启用的来源，执行抓取""" - sources = await get_enabled_sources() +async def run_fetch(sor=None): + """遍历所有启用的来源，执行抓取。sor 可选，由 DSPY 的 get_sor_context 提供。""" + sources = await get_enabled_sources(sor) now = now_str() results = {'total': len(sources), 'new_items': 0, 'errors': 0, 'details': []} @@ -124,18 +124,18 @@ async def run_fetch(): print(f'[hotspot] run_fetch error for {src.get("name",src["id"])}: {e}') traceback.print_exc() await write_fetch_log(src['id'], now, int((time.time() - t0) * 1000), - 'failed', 0, 0, 0, str(e)[:1000]) + 'failed', 0, 0, 0, str(e)[:1000], sor=sor) results['errors'] += 1 continue elapsed = int((time.time() - t0) * 1000) status = 'success' if code in (0, 200) else ('partial' if items else 'failed') - new_count, dup_count = await save_items(items, src['id'], now) - await update_source_last_fetch(src['id']) + new_count, dup_count = await save_items(items, src['id'], now, sor=sor) + await update_source_last_fetch(src['id'], sor=sor) await write_fetch_log(src['id'], now, elapsed, status, len(items), new_count, dup_count, - response_code=code, response_size_bytes=size) + response_code=code, response_size_bytes=size, sor=sor) results['new_items'] += new_count results['details'].append({ diff --git a/hotspot/init.py b/hotspot/init.py new file mode 100644 index 0000000..fab5267 --- /dev/null +++ b/hotspot/init.py @@ -0,0 +1,20 @@ +""" +热点雷达模块 — Sage 框架集成入口 + +在 sage/app/sage.py 中添加: + from hotspot.init import load_hotspot + 并在 init() 中调用: load_hotspot() +""" +from ahserver.serverenv import ServerEnv +from .engine import run_fetch +from .analysis import run_analysis +from .db import get_stats, get_enabled_sources + + +def load_hotspot(): + """注册热点雷达函数到 ServerEnv，供 DSPY 和 Jinja2 模板调用""" + env = ServerEnv() + env.run_fetch = run_fetch + env.run_analysis = run_analysis + env.get_stats = get_stats + env.get_enabled_sources = get_enabled_sources diff --git a/scripts/load_path.py b/scripts/load_path.py new file mode 100644 index 0000000..435d1e9 --- /dev/null +++ b/scripts/load_path.py @@ -0,0 +1,70 @@ +""" +热点雷达模块 — 独立权限注册脚本 +在 Sage 环境中执行: python scripts/load_path.py +""" +import os, sys, subprocess + +# 查找 Sage 根目录 (从当前模块位置推断) +def find_sage_root(): + for candidate in [ + os.path.expanduser('~/repos/sage'), + os.path.expanduser('~/sage'), + os.path.expanduser('~/py/sage'), + ]: + if os.path.isdir(os.path.join(candidate, 'py3', 'bin')): + return candidate + # Fallback: 通过导入路径查找 + try: + import load_path as _ # Sage 自己的 load_path.py + sage_root = os.path.dirname(os.path.abspath(_.__file__)) + return sage_root + except ImportError: + pass + print('ERROR: Sage root not found. Set SAGE_ROOT env var.') + sys.exit(1) + +SAGE_ROOT = os.environ.get('SAGE_ROOT', find_sage_root()) +PY = os.path.join(SAGE_ROOT, 'py3', 'bin', 'python') +SET_ROLE_PERM = os.path.join(SAGE_ROOT, 'set_role_perm.py') + +MOD = 'hotspot' + +PATHS_LOGINED = [ + f'/{MOD}', + f'/{MOD}/index.ui', + f'/{MOD}/stats.dspy', + f'/{MOD}/fetch_now.dspy', + f'/{MOD}/analyze.dspy', + # 6 张表的 CRUD + f'/{MOD}_source', + f'/{MOD}_source/%', + f'/{MOD}_schedule', + f'/{MOD}_schedule/%', + f'/{MOD}_fetch_log', + f'/{MOD}_fetch_log/%', + f'/{MOD}_item', + f'/{MOD}_item/%', + f'/{MOD}_analysis', + f'/{MOD}_analysis/%', + f'/{MOD}_alert', + f'/{MOD}_alert/%', +] + +def set_perm(role, path): + env = os.environ.copy() + env['SAGE_RBAC_DB'] = 'sage' + subprocess.run( + [PY, SET_ROLE_PERM, role, path], + cwd=SAGE_ROOT, capture_output=True, env=env + ) + +def main(): + total = 0 + for p in PATHS_LOGINED: + set_perm('logined', p) + total += 1 + print(f'Registered {total} hotspot paths') + print('Note: restart Sage for permissions to take effect') + +if __name__ == '__main__': + main() diff --git a/wwwroot/hotspot/analyze.dspy b/wwwroot/hotspot/analyze.dspy index 918270b..8e53463 100644 --- a/wwwroot/hotspot/analyze.dspy +++ b/wwwroot/hotspot/analyze.dspy @@ -1,2 +1,5 @@ -from hotspot.analysis import run_analysis -return await run_analysis() +import json +env = request._run_ns +async with get_sor_context(env, 'sage') as sor: + result = await env.run_analysis(sor) + return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/hotspot/fetch_now.dspy b/wwwroot/hotspot/fetch_now.dspy index 939d9f6..2087a8d 100644 --- a/wwwroot/hotspot/fetch_now.dspy +++ b/wwwroot/hotspot/fetch_now.dspy @@ -1,2 +1,5 @@ -from hotspot.engine import run_fetch -return await run_fetch() +import json +env = request._run_ns +async with get_sor_context(env, 'sage') as sor: + result = await env.run_fetch(sor) + return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/hotspot/stats.dspy b/wwwroot/hotspot/stats.dspy index cd064e5..828bbb7 100644 --- a/wwwroot/hotspot/stats.dspy +++ b/wwwroot/hotspot/stats.dspy @@ -1,2 +1,5 @@ -from hotspot.db import get_stats -return await get_stats() +import json +env = request._run_ns +async with get_sor_context(env, 'sage') as sor: + result = await env.get_stats(sor) + return json.dumps(result, ensure_ascii=False)