mao1/anchor_writer.py

#!/usr/bin/env python3
"""
Anchor Memory Writer
Writes memory to Obsidian Vault with YAML frontmatter and emotion tags.
Uses Qwen API (configured in Hermes) for summarization and tagging.
"""
import yaml
import json
import os
import urllib.request
import urllib.error
from datetime import datetime
import re

VAULT_DIR = "/Users/fyah/Documents/如梦初醒/memory/test/Anchor-Memory"
CONFIG_PATH = os.path.expanduser("~/.hermes/config.yaml")

def get_api_config():
    """Read API config from Hermes config.yaml"""
    with open(CONFIG_PATH, 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)
    
    providers = config.get('custom_providers', [])
    if not providers:
        raise ValueError("No custom providers found in config")
    
    # Find zhangsan provider first, then bailian
    zhangsan = None
    bailian = None
    for p in providers:
        if p.get('name') == 'zhangsan':
            zhangsan = p
        elif p.get('name') == 'bailian':
            bailian = p
            
    if zhangsan:
        return zhangsan['api_key'], zhangsan['base_url'], zhangsan.get('model', 'Qwen3.6-35B-A3B')
    elif bailian:
        return bailian['api_key'], bailian['base_url'], bailian.get('model', 'qwen3.5-plus')
    else:
        return providers[0]['api_key'], providers[0]['base_url'], providers[0].get('model', 'default')

def call_model(content):
    """Call Qwen API to generate summary, tags, and emotion scores"""
    api_key, base_url, model = get_api_config()
    
    # Simpler prompt, no response_format to avoid JSON parsing issues
    prompt = f"""Analyze this text. Output ONLY a valid JSON object. No markdown, no explanation. 
Keys: 
- summary (max 50 words, append URLs if external resources/GitHub mentioned)
- tags (list of EXACTLY 2 strings, format '#Category/ProjectName', e.g., ['#memory/zhang-san', '#llm/local-deploy'])
- valence (-1.0 to 1.0)
- arousal (0.0 to 1.0)
- timestamp (current time in YYYY-MM-DD HH:MM format)
IMPORTANT: If the text mentions "User", replace it with "Scintilla" in the summary.
Text: {content}"""
    
    data = json.dumps({
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.1
    }).encode('utf-8')

    req = urllib.request.Request(
        f"{base_url}/chat/completions",
        data=data,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}"
        }
    )
    
    try:
        with urllib.request.urlopen(req, timeout=30) as response:
            result = json.loads(response.read().decode('utf-8'))
            raw_content = result['choices'][0]['message']['content']
            
            # Clean up potential markdown code blocks
            raw_content = re.sub(r'^```json\s*', '', raw_content)
            raw_content = re.sub(r'\s*```$', '', raw_content)
            raw_content = raw_content.strip()
            
            return json.loads(raw_content)
    except Exception as e:
        print(f"[!] API Error: {e}")
        # Fallback if API fails
        return {
            "summary": content[:50],
            "tags": ["error", "fallback"],
            "valence": 0.0,
            "arousal": 0.0
        }

def save_memory(content):
    """Save memory to Obsidian Vault"""
    meta = call_model(content)
    now = datetime.now()
    filename = f"{now.strftime('%Y-%m-%d_%H%M')}.md"
    filepath = os.path.join(VAULT_DIR, filename)
    
    # Construct Markdown with YAML frontmatter
    ts = meta.get('timestamp', now.strftime('%Y-%m-%d %H:%M'))
    md_content = f"""---
title: {meta.get('summary', 'Untitled')[:30]}
date: {now.strftime('%Y-%m-%d %H:%M')}
tags: {meta.get('tags', [])}
emotion:
  valence: {meta.get('valence', 0.0)}
  arousal: {meta.get('arousal', 0.0)}
status: active
---

# {meta.get('summary', 'Untitled')}

> 📅 {ts}

{content}
"""
    with open(filepath, 'w', encoding='utf-8') as f:
        f.write(md_content)
    print(f"[+] Memory saved: {filepath}")

if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1:
        content = " ".join(sys.argv[1:])
        save_memory(content)
    else:
        print("Usage: python3 anchor_writer.py <content>")