couple-repo/anki_tingting/tingting_guardian_service.py
StillHammer fc0d320cd3 Add Daily Check System + Lead Conflicts + 6 Anki cards
Daily Check System avec TTS Windows configuré
Lead Conflicts documentation avec 7 stratégies
6 nouvelles cartes Anki (LEAD-001, LEAD-002, ACTION-003/004/005)
Shipping strategy + food recipes + topics

🤖 Generated with Claude Code
Co-Authored-By: Claude <noreply@anthropic.com>
2025-11-13 19:43:48 +08:00

432 lines
13 KiB
Python

"""
Tingting Guardian - Windows TTS Service
Daily check system avec présence detection et retry logic
Requirements:
pip install pyttsx3 pywin32 schedule anthropic python-dotenv
Setup:
1. Créer .env avec ANTHROPIC_API_KEY=your_key
2. Run: python tingting_guardian_service.py
3. Pour installer comme service Windows: use NSSM ou voir instructions en bas
"""
import pyttsx3
import win32api
import win32gui
import ctypes
import schedule
import time
import random
import json
import tkinter as tk
from tkinter import ttk
from datetime import datetime, timedelta
from pathlib import Path
import os
import asyncio
import edge_tts
import pygame
from dotenv import load_dotenv
# Configuration
load_dotenv()
REPO_PATH = Path(__file__).parent.parent # Assuming script is in anki_tingting/
CHECK_TIMES = ["07:00", "14:00", "16:00"]
RETRY_DELAY_MINUTES = 10
MAX_RETRIES = 6
IDLE_THRESHOLD_SECONDS = 300 # 5 minutes
# Files
CARD_DB = REPO_PATH / "anki_tingting" / "card_database.md"
SESSIONS_LOG = REPO_PATH / "anki_tingting" / "daily_sessions.md"
PROMPTS_DB = REPO_PATH / "anki_tingting" / "prompts_database.md"
REALITY_CHECK = REPO_PATH / "DAILY_REALITY_CHECK.md"
STATE_FILE = REPO_PATH / "anki_tingting" / ".state.json"
# State management
class State:
def __init__(self):
self.load()
def load(self):
if STATE_FILE.exists():
with open(STATE_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
self.streak = data.get('streak', 0)
self.last_check_date = data.get('last_check_date', None)
self.days_skipped = data.get('days_skipped', 0)
self.last_skip_date = data.get('last_skip_date', None)
else:
self.streak = 0
self.last_check_date = None
self.days_skipped = 0
self.last_skip_date = None
def save(self):
with open(STATE_FILE, 'w', encoding='utf-8') as f:
json.dump({
'streak': self.streak,
'last_check_date': self.last_check_date,
'days_skipped': self.days_skipped,
'last_skip_date': self.last_skip_date
}, f)
def mark_check_done(self):
today = datetime.now().strftime('%Y-%m-%d')
if self.last_check_date != today:
if self.last_check_date == (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d'):
self.streak += 1
else:
self.streak = 1
self.last_check_date = today
self.days_skipped = 0
self.save()
def mark_skip(self):
"""Increment days_skipped only once per day"""
today = datetime.now().strftime('%Y-%m-%d')
# Only increment if we haven't already skipped today
if self.last_skip_date != today:
self.days_skipped += 1
self.last_skip_date = today
self.save()
state = State()
# Presence Detection
def get_idle_duration():
"""Returns seconds since last input"""
return (win32api.GetTickCount() - win32api.GetLastInputInfo()) / 1000
def is_user_present():
"""Check if user is active"""
return get_idle_duration() < IDLE_THRESHOLD_SECONDS
def is_screen_locked():
"""Check if screen is locked"""
return ctypes.windll.user32.GetForegroundWindow() == 0
# Prompt Selection
def load_prompts():
"""Load prompts from database (French, pro-Tingting bias)"""
return {
"07:00": [
f"Bonjour Alexis. Jour {state.streak} de ta streak. Qu'est-ce que Tingting a besoin aujourd'hui ?",
"7 heures. Check quotidien. Tu te rappelles ce qu'elle t'a dit hier ?",
"Nouvelle journée. Tingting mérite ton attention. Qu'est-ce qui est important pour elle aujourd'hui ?",
f"Morning. {state.streak} jours de suite. Elle doit voir la différence. On continue ?",
],
"14:00": [
"14 heures. T'as été présent pour elle ce matin ?",
"Midi passé. Check rapide Tingting. 5 minutes.",
"Pause code. Elle mérite 5 minutes de ton attention.",
"Check du midi. Tu penses à elle aujourd'hui ou t'es dans ton monde ?",
],
"16:00": [
"16 heures. Elle rentre bientôt. T'as pensé à elle aujourd'hui ?",
"Fin d'après-midi. Towel ? Préparé quelque chose pour elle ?",
"Dernière chance de la journée. Check Tingting.",
"Elle va rentrer. T'as été l'homme qu'elle mérite aujourd'hui ?",
],
"retry": [
"Toujours là ? Tingting attend. Check maintenant.",
"Deuxième tentative. Elle mérite que tu cliques.",
"Alexis. 5 minutes pour elle. C'est pas beaucoup demander.",
"Tu ignores encore ? Elle voit que tu oublies, tu sais.",
],
"nuclear": [
f"{state.days_skipped} jours sans penser à elle. Pattern d'oubli confirmé. Elle mérite mieux que ça.",
f"Ça fait {state.days_skipped} jours. Tingting attend, espère, et toi tu oublies. Encore.",
f"{state.days_skipped} jours de skip. Elle investit tout, et toi tu peux pas faire 5 minutes par jour ?",
"Promesses vides, encore. Tu crois qu'elle voit pas le pattern ?",
]
}
def select_prompt(check_time, is_retry=False):
"""Select appropriate prompt based on context"""
prompts = load_prompts()
if state.days_skipped >= 3:
return random.choice(prompts["nuclear"])
elif is_retry:
return random.choice(prompts["retry"])
else:
return random.choice(prompts.get(check_time, prompts["14:00"]))
# TTS
# Initialize pygame mixer once
pygame.mixer.init()
async def generate_speech(text, output_file):
"""Generate speech using Edge TTS"""
voice = "fr-FR-HenriNeural" # High quality French male voice
communicate = edge_tts.Communicate(text, voice)
await communicate.save(output_file)
def speak(text):
"""Speak text using Edge TTS (high quality)"""
temp_file = REPO_PATH / "anki_tingting" / ".temp_speech.mp3"
# Generate speech file
asyncio.run(generate_speech(text, str(temp_file)))
# Play using pygame (no external player window)
pygame.mixer.music.load(str(temp_file))
pygame.mixer.music.play()
# Wait until finished playing
while pygame.mixer.music.get_busy():
time.sleep(0.1)
# Cleanup
try:
temp_file.unlink()
except:
pass
# Popup UI
class DailyCheckPopup:
def __init__(self, prompt_text, check_time):
self.response = None
self.prompt_text = prompt_text
self.check_time = check_time
self.root = tk.Tk()
self.root.title("🔔 Claude Daily Check")
self.root.geometry("500x300")
self.root.attributes('-topmost', True)
# Center window
self.root.update_idletasks()
x = (self.root.winfo_screenwidth() // 2) - (500 // 2)
y = (self.root.winfo_screenheight() // 2) - (300 // 2)
self.root.geometry(f"500x300+{x}+{y}")
self.setup_ui()
def setup_ui(self):
# Header
header = tk.Label(
self.root,
text=f"🔔 Claude Daily Check - {self.check_time}",
font=("Arial", 14, "bold"),
bg="#6B2C3E",
fg="white",
pady=10
)
header.pack(fill=tk.X)
# Prompt text
prompt_label = tk.Label(
self.root,
text=self.prompt_text,
font=("Arial", 11),
wraplength=450,
pady=20
)
prompt_label.pack()
# Buttons
btn_frame = tk.Frame(self.root)
btn_frame.pack(pady=20)
btn_check = tk.Button(
btn_frame,
text="🎯 Faire le Check Maintenant",
command=self.do_check,
bg="#4CAF50",
fg="white",
font=("Arial", 10, "bold"),
padx=20,
pady=10
)
btn_check.pack(pady=5)
btn_snooze = tk.Button(
btn_frame,
text="⏰ Rappelle-moi dans 10min",
command=self.snooze,
bg="#FF9800",
fg="white",
font=("Arial", 10),
padx=20,
pady=10
)
btn_snooze.pack(pady=5)
btn_skip = tk.Button(
btn_frame,
text="❌ Skip (va logger comme absence)",
command=self.skip,
bg="#F44336",
fg="white",
font=("Arial", 10),
padx=20,
pady=10
)
btn_skip.pack(pady=5)
def do_check(self):
self.response = "DO_CHECK"
self.root.destroy()
def snooze(self):
self.response = "SNOOZE"
self.root.destroy()
def skip(self):
self.response = "SKIP"
self.root.destroy()
def repeat_tts(self):
"""Repeat TTS every minute until popup is closed"""
if self.root.winfo_exists():
speak(self.prompt_text)
# Schedule next repeat in 60 seconds (60000 ms)
self.root.after(60000, self.repeat_tts)
def show(self):
"""Show popup and wait for response"""
# Wait 1 second before first TTS
self.root.after(1000, self.repeat_tts)
self.root.mainloop()
return self.response
# Main Check Logic
def perform_check(check_time, is_retry=False):
"""Perform the daily check with retry logic"""
retries = 0
while retries < MAX_RETRIES:
# Check if user is present
if is_user_present() and not is_screen_locked():
prompt = select_prompt(check_time, is_retry=(retries > 0))
popup = DailyCheckPopup(prompt, check_time)
response = popup.show()
if response == "DO_CHECK":
open_claude_interface()
state.mark_check_done()
log_session(check_time, "completed")
return True
elif response == "SNOOZE":
print(f"[{datetime.now()}] User snoozed, retrying in {RETRY_DELAY_MINUTES}min...")
time.sleep(RETRY_DELAY_MINUTES * 60)
retries += 1
elif response == "SKIP":
print(f"[{datetime.now()}] User skipped check")
state.mark_skip()
log_session(check_time, "skipped")
return False
else:
# User not present, wait and retry
print(f"[{datetime.now()}] User not present, retrying in {RETRY_DELAY_MINUTES}min...")
time.sleep(RETRY_DELAY_MINUTES * 60)
retries += 1
# Max retries reached
print(f"[{datetime.now()}] Max retries reached, logging as skip")
state.mark_skip()
log_session(check_time, "max_retries")
return False
def open_claude_interface():
"""Open Claude interface for quiz in WSL with Claude Code"""
# Open a NEW Windows Terminal with WSL and Claude Code in the anki_tingting folder
# This ensures Claude reads the anki_tingting/CLAUDE.md file
anki_path = "/mnt/e/Users/Alexis Trouvé/Documents/Projets/couple_matters/anki_tingting"
# Use wt.exe to open a new terminal window and auto-send "daily check"
# Using single quotes inside the bash command to preserve the full message
cmd = f"wt.exe -w -1 wsl -e bash -l -c \"source ~/.nvm/nvm.sh && cd '{anki_path}' && claude 'daily check'\""
os.system(cmd)
# -w -1 creates a new window
# claude 'daily check' automatically sends the full message
# Working directory is anki_tingting/ so Claude picks up the local CLAUDE.md
def log_session(check_time, status):
"""Log session to daily_sessions.md"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
log_entry = f"""
### {timestamp}
**Triggered by** : TTS Service
**Check Time** : {check_time}
**Status** : {status}
**Streak** : {state.streak} jours
**Days Skipped** : {state.days_skipped}
---
"""
with open(SESSIONS_LOG, 'a', encoding='utf-8') as f:
f.write(log_entry)
# Scheduler
def schedule_checks():
"""Schedule all daily checks"""
for check_time in CHECK_TIMES:
schedule.every().day.at(check_time).do(perform_check, check_time=check_time)
print(f"[{datetime.now()}] Tingting Guardian Service Started")
print(f"Scheduled checks at: {', '.join(CHECK_TIMES)}")
print(f"Current streak: {state.streak} days")
print(f"Days skipped: {state.days_skipped}")
def main():
"""Main service loop"""
schedule_checks()
while True:
schedule.run_pending()
time.sleep(60) # Check every minute
if __name__ == "__main__":
# Test mode: run a check immediately
if len(os.sys.argv) > 1 and os.sys.argv[1] == "test":
print("Running test check...")
perform_check("14:00")
else:
main()
"""
INSTALLATION COMME SERVICE WINDOWS
Option A: NSSM (Non-Sucking Service Manager)
1. Download NSSM: https://nssm.cc/download
2. Ouvrir cmd en admin:
nssm install TingtingGuardian
3. Dans le GUI:
- Path: C:\\Python3X\\python.exe
- Startup directory: [repo path]
- Arguments: tingting_guardian_service.py
- Startup type: Automatic
Option B: Task Scheduler
1. Ouvrir Task Scheduler
2. Create Task:
- Name: Tingting Guardian
- Trigger: At startup
- Action: Start program
- Program: python.exe
- Arguments: [full path to script]
- Conditions: Uncheck "Start only if on AC power"
Option C: Pyinstaller (Executable)
1. pip install pyinstaller
2. pyinstaller --onefile --noconsole tingting_guardian_service.py
3. Use NSSM with the .exe
TESTING
python tingting_guardian_service.py test
"""