xorbitlab
/
dom
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

This commit is contained in:
root 2025-08-30 15:55:23 +00:00
commit 7d3770dba3
11 changed files with 794 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

207
.gitignore vendored Normal file
View File

@ -0,0 +1,207 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be added to the global gitignore or merged into this project gitignore. For a PyCharm
# project, it is recommended to include the following files:
# .idea/
# *.iml
# *.ipr
# *.iws
.idea/
*.iml
*.ipr
*.iws
# VS Code
.vscode/
*.code-workspace
# Claude AI
.claude/
# OS generated files
.DS_Store
.DS_Store?
._*
.Spotlight-V100
.Trashes
ehthumbs.db
Thumbs.db
# Temporary files
*.tmp
*.temp
*.swp
*.swo
*~
# Logs
*.log
logs/
# Local configuration files
config.local.json
*.local.json
.env.local
# Database files
*.db
*.sqlite
*.sqlite3
# Backup files
*.bak
*.backup

19
CLAUDE.md Normal file
View File

@ -0,0 +1,19 @@
Project Description:
This is a python based project. keep it lightweighted
Features:
1. the app can invoke ollama client to send prompt and get response.
2. send the llm response to 3rd party rest apis, this apis can push notification to ios/andorid. the apis are not in the scope
3. the prompt can have different template, each time it will pick the template ramdonly.
4. each template have some placeholders, for example, fill in the date, time of the real time
5. the invoke ollama and send notifications will be triggered at random intervals(around 45 minutes, it can be 3 or 5 minutes, can be 2 or 3 hours)
6. the invoke will be set with a slient time, for example every day 12pm to 8am.
7. the silent time, ollama api endpoint, etc, can be configurable
8. Notification APIs:
regarding notification api, there are 2 apis to push:
bark:
https://bark.day.app/#/tutorial?id=%e5%8f%91%e9%80%81%e6%8e%a8%e9%80%81
ntfy:
https://docs.ntfy.sh/publish/

14
config.json Normal file
View File

@ -0,0 +1,14 @@
{
"ollama_endpoint": "http://localhost:11434",
"ollama_model": "llama2",
"silent_start": "20:00",
"silent_end": "12:00",
"min_interval": 3,
"max_interval": 180,
"bark_api_url": "",
"bark_device_key": "",
"ntfy_api_url": "",
"ntfy_topic": "",
"ntfy_access_token": "",
"templates_dir": "templates"
}

47
config.py Normal file
View File

@ -0,0 +1,47 @@
import json
from pathlib import Path
from typing import Dict, Any
from dataclasses import dataclass, asdict
@dataclass
class Config:
"""Configuration management for the notification system."""
ollama_endpoint: str = "http://localhost:11434"
ollama_model: str = "llama2"
# Silent time configuration (12pm to 8am)
silent_start: str = "20:00"
silent_end: str = "12:00"
# Interval configuration (in minutes)
min_interval: int = 3
max_interval: int = 180
# Bark notification service
bark_api_url: str = ""
bark_device_key: str = ""
# Ntfy notification service
ntfy_api_url: str = ""
ntfy_topic: str = ""
ntfy_access_token: str = ""
# Template settings
templates_dir: str = "templates"
def __post_init__(self):
"""Load configuration from file if exists."""
config_file = Path("config.json")
if config_file.exists():
with open(config_file, 'r') as f:
data = json.load(f)
for key, value in data.items():
if hasattr(self, key):
setattr(self, key, value)
def save(self):
"""Save current configuration to file."""
config_file = Path("config.json")
with open(config_file, 'w') as f:
json.dump(asdict(self), f, indent=2)

75
config_cli.py Executable file
View File

@ -0,0 +1,75 @@
#!/usr/bin/env python3
"""CLI tool for managing configuration."""
import argparse
import asyncio
from config import Config
async def main():
parser = argparse.ArgumentParser(description="Ollama Notification System Configuration")
parser.add_argument("--set-ollama-endpoint", help="Set Ollama endpoint URL")
parser.add_argument("--set-ollama-model", help="Set Ollama model name")
parser.add_argument("--set-silent-time", nargs=2, metavar=("START", "END"),
help="Set silent time range (e.g., 20:00 12:00)")
parser.add_argument("--set-interval-range", nargs=2, type=int, metavar=("MIN", "MAX"),
help="Set notification interval range in minutes")
parser.add_argument("--set-bark-url", help="Set Bark API URL")
parser.add_argument("--set-bark-key", help="Set Bark device key")
parser.add_argument("--set-ntfy-url", help="Set Ntfy API URL")
parser.add_argument("--set-ntfy-topic", help="Set Ntfy topic")
parser.add_argument("--set-ntfy-token", help="Set Ntfy access token")
parser.add_argument("--show", action="store_true", help="Show current configuration")
args = parser.parse_args()
config = Config()
if args.set_ollama_endpoint:
config.ollama_endpoint = args.set_ollama_endpoint
if args.set_ollama_model:
config.ollama_model = args.set_ollama_model
if args.set_silent_time:
config.silent_start = args.set_silent_time[0]
config.silent_end = args.set_silent_time[1]
if args.set_interval_range:
config.min_interval = args.set_interval_range[0]
config.max_interval = args.set_interval_range[1]
if args.set_bark_url:
config.bark_api_url = args.set_bark_url
if args.set_bark_key:
config.bark_device_key = args.set_bark_key
if args.set_ntfy_url:
config.ntfy_api_url = args.set_ntfy_url
if args.set_ntfy_topic:
config.ntfy_topic = args.set_ntfy_topic
if args.set_ntfy_token:
config.ntfy_access_token = args.set_ntfy_token
if any([args.set_ollama_endpoint, args.set_ollama_model, args.set_silent_time,
args.set_interval_range, args.set_bark_url, args.set_bark_key,
args.set_ntfy_url, args.set_ntfy_topic, args.set_ntfy_token]):
config.save()
print("Configuration saved successfully!")
if args.show or not any(vars(args).values()):
print("Current Configuration:")
print(f" Ollama Endpoint: {config.ollama_endpoint}")
print(f" Ollama Model: {config.ollama_model}")
print(f" Silent Time: {config.silent_start} - {config.silent_end}")
print(f" Interval Range: {config.min_interval} - {config.max_interval} minutes")
print(f" Bark API URL: {config.bark_api_url or 'Not set'}")
print(f" Bark Device Key: {'***' if config.bark_device_key else 'Not set'}")
print(f" Ntfy API URL: {config.ntfy_api_url or 'Not set'}")
print(f" Ntfy Topic: {config.ntfy_topic or 'Not set'}")
print(f" Ntfy Token: {'***' if config.ntfy_access_token else 'Not set'}")
if __name__ == "__main__":
asyncio.run(main())

23
main.py Normal file
View File

@ -0,0 +1,23 @@
import asyncio
import logging
from pathlib import Path
from scheduler import NotificationScheduler
from config import Config
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def main():
"""Main entry point for the notification system."""
config = Config()
scheduler = NotificationScheduler(config)
try:
await scheduler.start()
except KeyboardInterrupt:
logger.info("Shutting down...")
except Exception as e:
logger.error(f"Error in main: {e}")
if __name__ == "__main__":
asyncio.run(main())

119
notification_client.py Normal file
View File

@ -0,0 +1,119 @@
import aiohttp
import logging
from typing import Dict, Any, Optional, List
from config import Config
logger = logging.getLogger(__name__)
class NotificationClient:
"""Client for sending notifications to both Bark and Ntfy services."""
def __init__(self, config: Config):
self.config = config
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def send_notification(self, message: str, title: str = "AI Thought") -> bool:
"""Send notification to both Bark and Ntfy services."""
results = []
# Send to Bark if configured
if self.config.bark_api_url and self.config.bark_device_key:
bark_success = await self._send_bark_notification(message, title)
results.append(bark_success)
# Send to Ntfy if configured
if self.config.ntfy_api_url and self.config.ntfy_topic:
ntfy_success = await self._send_ntfy_notification(message, title)
results.append(ntfy_success)
# Return True if at least one service succeeded
return any(results) if results else True
async def _send_bark_notification(self, message: str, title: str) -> bool:
"""Send notification to Bark service."""
if not self.session:
self.session = aiohttp.ClientSession()
# Bark API format: https://api.day.app/{device_key}/{title}/{body}
url = f"{self.config.bark_api_url}/{self.config.bark_device_key}/{title}/{message}"
try:
async with self.session.get(url) as response:
if response.status == 200:
data = await response.json()
if data.get("code") == 200:
logger.info("Bark notification sent successfully")
return True
else:
logger.error(f"Bark API error: {data}")
return False
else:
logger.error(f"Bark HTTP error: {response.status}")
return False
except Exception as e:
logger.error(f"Error sending Bark notification: {e}")
return False
async def _send_ntfy_notification(self, message: str, title: str) -> bool:
"""Send notification to Ntfy service."""
if not self.session:
self.session = aiohttp.ClientSession()
url = f"{self.config.ntfy_api_url}/{self.config.ntfy_topic}"
headers = {
"Title": title,
"Content-Type": "text/plain",
"User-Agent": "Ollama-Notification-System/1.0"
}
if self.config.ntfy_access_token:
headers["Authorization"] = f"Bearer {self.config.ntfy_access_token}"
try:
async with self.session.post(url, data=message, headers=headers) as response:
if response.status == 200:
logger.info("Ntfy notification sent successfully")
return True
else:
logger.error(f"Ntfy HTTP error: {response.status}")
return False
except Exception as e:
logger.error(f"Error sending Ntfy notification: {e}")
return False
async def test_connections(self) -> Dict[str, bool]:
"""Test connections to both notification services."""
results = {}
# Test Bark
if self.config.bark_api_url:
try:
async with self.session.get(self.config.bark_api_url, timeout=aiohttp.ClientTimeout(total=10)) as response:
results["bark"] = response.status < 400
except Exception:
results["bark"] = False
else:
results["bark"] = None
# Test Ntfy
if self.config.ntfy_api_url:
try:
async with self.session.get(self.config.ntfy_api_url, timeout=aiohttp.ClientTimeout(total=10)) as response:
results["ntfy"] = response.status < 400
except Exception:
results["ntfy"] = False
else:
results["ntfy"] = None
return results

58
ollama_client.py Normal file
View File

@ -0,0 +1,58 @@
import aiohttp
import json
import logging
from typing import Dict, Any, Optional
from config import Config
logger = logging.getLogger(__name__)
class OllamaClient:
"""Client for interacting with Ollama API."""
def __init__(self, config: Config):
self.config = config
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def generate_response(self, prompt: str) -> Optional[str]:
"""Generate a response from Ollama for the given prompt."""
if not self.session:
self.session = aiohttp.ClientSession()
url = f"{self.config.ollama_endpoint}/api/generate"
payload = {
"model": self.config.ollama_model,
"prompt": prompt,
"stream": False
}
try:
async with self.session.post(url, json=payload) as response:
if response.status == 200:
data = await response.json()
return data.get("response", "").strip()
else:
logger.error(f"Ollama API error: {response.status}")
return None
except Exception as e:
logger.error(f"Error calling Ollama API: {e}")
return None
async def check_health(self) -> bool:
"""Check if Ollama service is available."""
if not self.session:
self.session = aiohttp.ClientSession()
try:
async with self.session.get(f"{self.config.ollama_endpoint}/api/tags") as response:
return response.status == 200
except Exception as e:
logger.error(f"Ollama health check failed: {e}")
return False

2
requirements.txt Normal file
View File

@ -0,0 +1,2 @@
aiohttp>=3.8.0
asyncio

108
scheduler.py Normal file
View File

@ -0,0 +1,108 @@
import asyncio
import random
import logging
from datetime import datetime, time, timedelta
from typing import Optional
from config import Config
from ollama_client import OllamaClient
from template_manager import TemplateManager
from notification_client import NotificationClient
logger = logging.getLogger(__name__)
class NotificationScheduler:
"""Handles scheduling of notifications with random intervals and silent time."""
def __init__(self, config: Config):
self.config = config
self.template_manager = TemplateManager(config.templates_dir)
self.notification_client = NotificationClient(config)
self.running = False
def _is_silent_time(self) -> bool:
"""Check if current time is within silent hours."""
current_time = datetime.now().time()
# Parse silent time configuration
silent_start = datetime.strptime(self.config.silent_start, "%H:%M").time()
silent_end = datetime.strptime(self.config.silent_end, "%H:%M").time()
# Handle overnight silent period (e.g., 20:00 to 12:00)
if silent_start >= silent_end:
return current_time >= silent_start or current_time <= silent_end
else:
return silent_start <= current_time <= silent_end
def _get_next_interval(self) -> int:
"""Get random interval in seconds between min and max."""
minutes = random.randint(self.config.min_interval, self.config.max_interval)
return minutes * 60
async def _send_notification(self) -> bool:
"""Send a notification with Ollama response."""
try:
async with OllamaClient(self.config) as client:
# Check if Ollama is available
if not await client.check_health():
logger.error("Ollama service is not available")
return False
# Get random prompt
prompt = self.template_manager.get_random_prompt()
logger.info(f"Using prompt: {prompt}")
# Get response from Ollama
response = await client.generate_response(prompt)
if not response:
logger.error("Failed to get response from Ollama")
return False
# Send notification
success = await self.notification_client.send_notification(response)
if success:
logger.info("Notification sent successfully")
else:
logger.error("Failed to send notification")
return success
except Exception as e:
logger.error(f"Error sending notification: {e}")
return False
async def start(self):
"""Start the notification scheduler."""
self.running = True
logger.info("Starting notification scheduler...")
while self.running:
try:
# Check if it's silent time
if self._is_silent_time():
logger.info("Silent time - skipping notification")
# Wait 30 minutes before checking again
await asyncio.sleep(1800)
continue
# Send notification
await self._send_notification()
# Calculate next interval
interval = self._get_next_interval()
next_time = datetime.now() + timedelta(seconds=interval)
logger.info(f"Next notification scheduled for: {next_time.strftime('%Y-%m-%d %H:%M:%S')}")
# Wait for next interval
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info("Scheduler cancelled")
break
except Exception as e:
logger.error(f"Error in scheduler: {e}")
# Wait 5 minutes on error
await asyncio.sleep(300)
def stop(self):
"""Stop the notification scheduler."""
self.running = False

122
template_manager.py Normal file
View File

@ -0,0 +1,122 @@
import json
import random
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Any
import logging
logger = logging.getLogger(__name__)
class TemplateManager:
"""Manages prompt templates with placeholders and random selection."""
def __init__(self, templates_dir: str = "templates"):
self.templates_dir = Path(templates_dir)
self.templates_dir.mkdir(exist_ok=True)
self.templates = []
self._load_templates()
def _load_templates(self):
"""Load templates from JSON files in the templates directory."""
if not self.templates_dir.exists():
self._create_default_templates()
for template_file in self.templates_dir.glob("*.json"):
try:
with open(template_file, 'r') as f:
template_data = json.load(f)
if isinstance(template_data, list):
self.templates.extend(template_data)
else:
self.templates.append(template_data)
except Exception as e:
logger.error(f"Error loading template {template_file}: {e}")
if not self.templates:
self._create_default_templates()
def _create_default_templates(self):
"""Create default templates if none exist."""
default_templates = [
{
"name": "Daily Reflection",
"prompt": "What are your thoughts about {date} at {time}?",
"description": "A simple daily reflection prompt"
},
{
"name": "Mindfulness Check",
"prompt": "How are you feeling right now on this {day} at {time}?",
"description": "A mindfulness check-in prompt"
},
{
"name": "Gratitude Prompt",
"prompt": "What are three things you're grateful for on {date} during this {time_of_day}?",
"description": "A gratitude reflection prompt"
}
]
template_file = self.templates_dir / "default.json"
with open(template_file, 'w') as f:
json.dump(default_templates, f, indent=2)
self.templates.extend(default_templates)
def _get_time_of_day(self, hour: int) -> str:
"""Return time of day based on hour."""
if 5 <= hour < 12:
return "morning"
elif 12 <= hour < 17:
return "afternoon"
elif 17 <= hour < 21:
return "evening"
else:
return "night"
def _fill_placeholders(self, template: str) -> str:
"""Fill placeholders in template with current values."""
now = datetime.now()
placeholders = {
"date": now.strftime("%Y-%m-%d"),
"time": now.strftime("%H:%M"),
"day": now.strftime("%A"),
"time_of_day": self._get_time_of_day(now.hour),
"weekday": now.strftime("%A"),
"month": now.strftime("%B"),
"day_of_month": str(now.day)
}
for placeholder, value in placeholders.items():
template = template.replace(f"{{{placeholder}}}", value)
return template
def get_random_prompt(self) -> str:
"""Get a random prompt template with placeholders filled."""
if not self.templates:
logger.warning("No templates available")
return "What are your thoughts right now?"
template = random.choice(self.templates)
prompt = template.get("prompt", "")
return self._fill_placeholders(prompt)
def add_template(self, name: str, prompt: str, description: str = ""):
"""Add a new template to the system."""
new_template = {
"name": name,
"prompt": prompt,
"description": description
}
self.templates.append(new_template)
# Save to file
template_file = self.templates_dir / "custom.json"
custom_templates = []
if template_file.exists():
with open(template_file, 'r') as f:
custom_templates = json.load(f)
custom_templates.append(new_template)
with open(template_file, 'w') as f:
json.dump(custom_templates, f, indent=2)