53 lines
2.3 KiB
Python
53 lines
2.3 KiB
Python
patch_code = '''
|
|
|
|
# ==================== 统计 API ====================
|
|
|
|
@app.get("/stats/overview")
|
|
async def get_stats_overview(db: Session = Depends(get_db)):
|
|
from sqlalchemy import func as sqlfunc
|
|
total_agents = db.query(sqlfunc.count(Agent.id)).scalar()
|
|
running_agents = db.query(sqlfunc.count(Agent.id)).filter(Agent.status == AgentStatus.RUNNING).scalar()
|
|
stopped_agents = db.query(sqlfunc.count(Agent.id)).filter(Agent.status == AgentStatus.STOPPED).scalar()
|
|
total_templates = db.query(sqlfunc.count(Agent.template_id.distinct())).scalar()
|
|
return {"total_agents": total_agents, "running_agents": running_agents, "stopped_agents": stopped_agents, "total_templates": total_templates}
|
|
|
|
@app.get("/stats/by-template")
|
|
async def get_stats_by_template(db: Session = Depends(get_db)):
|
|
from sqlalchemy import func as sqlfunc
|
|
result = db.query(
|
|
Template.name,
|
|
sqlfunc.count(Agent.id).label("agent_count"),
|
|
sqlfunc.sum(Agent.current_replicas).label("total_replicas")
|
|
).outerjoin(Template, Agent.template_id == Template.id).group_by(Template.name).all()
|
|
return [{"template_name": r[0], "agent_count": r[1], "total_replicas": r[2] or 0} for r in result]
|
|
|
|
@app.get("/stats/by-owner")
|
|
async def get_stats_by_owner(db: Session = Depends(get_db)):
|
|
from sqlalchemy import func as sqlfunc
|
|
agents = db.query(Agent).all()
|
|
owner_map = {}
|
|
for a in agents:
|
|
oid = a.owner_id
|
|
if oid not in owner_map:
|
|
owner_map[oid] = {"owner_id": oid, "agent_count": 0, "total_replicas": 0, "platform_agents": 0, "custom_agents": 0}
|
|
owner_map[oid]["agent_count"] += 1
|
|
owner_map[oid]["total_replicas"] += a.current_replicas or 0
|
|
if a.agent_type == AgentType.PLATFORM:
|
|
owner_map[oid]["platform_agents"] += 1
|
|
else:
|
|
owner_map[oid]["custom_agents"] += 1
|
|
return list(owner_map.values())
|
|
|
|
'''
|
|
|
|
import re
|
|
path = '/home/xiaohei/agent_management/app.py'
|
|
with open(path, 'r') as f:
|
|
content = f.read()
|
|
|
|
content = re.sub(r'\n# ==================== 统计 API ====================.*?(?=\nif __name__)', '', content, flags=re.DOTALL)
|
|
content = content.replace('if __name__ == "__main__":', patch_code + 'if __name__ == "__main__":')
|
|
with open(path, 'w') as f:
|
|
f.write(content)
|
|
print('patched ok')
|