yumoqing/hotspot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Sage-compliant DSPY — no imports, env delegation, init.py, scripts/load_path.py

Browse Source 
- 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
This commit is contained in:
yumoqing 2026-08-01 16:49:49 +08:00
parent b9bd705b43
commit 7bb9bfc97e
9 changed files with 213 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
hotspot/__init__.py
View File

@ -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']

63
hotspot/analysis.py
View File

@ -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

89
hotspot/db.py
View File

@ -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:

14
hotspot/engine.py
View File

@ -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({

20
hotspot/init.py Normal file
View File

@ -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

70
scripts/load_path.py Normal file
View File

@ -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()

7
wwwroot/hotspot/analyze.dspy
View File

@ -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)

7
wwwroot/hotspot/fetch_now.dspy
View File

@ -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)

7
wwwroot/hotspot/stats.dspy
View File

@ -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)