Compare commits

..

10 Commits

Author SHA1 Message Date
Drew Cain 7adf6791e9 fix: #190 MCP Hub tool validation
Signed-off-by: Drew Cain <groksrc@gmail.com>
2025-06-30 23:12:28 -05:00
github-actions[bot] 1dc66bec7a 📊 Daily metrics update - 2025-07-01 2025-07-01 01:24:38 +00:00
nellins 42d97504a3 Update daily_report.py
Signed-off-by: nellins <drewnellins@gmail.com>
2025-06-30 20:24:08 -05:00
nellins 6112e185fd Create daily_report.py
Signed-off-by: nellins <drewnellins@gmail.com>
2025-06-30 20:20:42 -05:00
nellins 2e758414da Create requirements.txt
Signed-off-by: nellins <drewnellins@gmail.com>
2025-06-30 20:19:22 -05:00
nellins 3ec9ca234e Create daily-traction.yml
Signed-off-by: nellins <drewnellins@gmail.com>
2025-06-30 20:10:49 -05:00
nellins 9dec7e98a8 Create daily_report.py
Signed-off-by: nellins <drewnellins@gmail.com>
2025-06-30 20:05:32 -05:00
nellins 338d225a7e Create requirements.txt
Signed-off-by: nellins <drewnellins@gmail.com>
2025-06-30 20:04:04 -05:00
nellins 211b760522 Create daily-traction.yml
Signed-off-by: nellins <drewnellins@gmail.com>
2025-06-30 20:03:01 -05:00
Drew Cain 39f811f8b5 Update README.md
Added Homebrew instructions to README.md

Signed-off-by: Drew Cain <groksrc@users.noreply.github.com>
2025-06-26 21:51:14 -05:00
15 changed files with 899 additions and 22 deletions
+28
View File
@@ -0,0 +1,28 @@
name: Daily Traction Report
on:
schedule:
- cron: '0 14 * * 1-5'
workflow_dispatch:
jobs:
generate-report:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.9'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r scripts/requirements.txt
- name: Generate Daily Traction Report
run: python scripts/daily_report.py
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
DISCORD_WEBHOOK: ${{ secrets.DISCORD_WEBHOOK }}
REDDIT_CLIENT_ID: ${{ secrets.REDDIT_CLIENT_ID }}
REDDIT_SECRET: ${{ secrets.REDDIT_SECRET }}
YOUTUBE_API_KEY: ${{ secrets.YOUTUBE_API_KEY }}
+4
View File
@@ -33,6 +33,10 @@ https://github.com/user-attachments/assets/a55d8238-8dd0-454a-be4c-8860dbbd0ddc
# Install with uv (recommended)
uv tool install basic-memory
# or with Homebrew
brew tap basicmachines-co/basic-memory
brew install basic-memory
# Configure Claude Desktop (edit ~/Library/Application Support/Claude/claude_desktop_config.json)
# Add this to your config:
{
+44
View File
@@ -0,0 +1,44 @@
{
"date": "2025-07-01T01:24:37.731009",
"metrics": {
"github": {
"stars": 1009,
"forks": 73,
"watchers": 1009,
"open_issues": 31,
"traffic_views": 0,
"traffic_unique": 0,
"recent_issues": 0
},
"reddit": {
"total_mentions": 25,
"subreddit_members": 43,
"top_posts": [
{
"title": "Today's banning is the largest since Affinity, and...",
"score": 892,
"subreddit": "magicTCG",
"num_comments": 127
},
{
"title": "Basic Memory - Backed by Research*",
"score": 1,
"subreddit": "basicmemory",
"num_comments": 0
},
{
"title": "n8n Full Course \u2699\ufe0f Create Our Second Flow \ud83d\udc49 Trigge...",
"score": 1,
"subreddit": "Tech_UpSkill",
"num_comments": 0
}
],
"hot_discussions": []
},
"youtube": {
"subscribers": 45,
"total_views": 1093,
"video_count": 6
}
}
}
+4 -4
View File
@@ -129,10 +129,10 @@ When using Docker volumes, you'll need to configure projects to point to your mo
2. **Add a project for your mounted volume:**
```bash
# If you mounted /path/to/your/vault to /app/data
docker exec basic-memory-server basic-memory project create my-vault /app/data
docker exec basic-memory-server basic-memory project add my-vault /app/data
# Set it as default
docker exec basic-memory-server basic-memory project set-default my-vault
docker exec basic-memory-server basic-memory project default my-vault
```
3. **Sync the new project:**
@@ -151,10 +151,10 @@ volumes:
Then configure it:
```bash
# Create project pointing to mounted vault
docker exec basic-memory-server basic-memory project create obsidian /app/data
docker exec basic-memory-server basic-memory project add obsidian /app/data
# Set as default
docker exec basic-memory-server basic-memory project set-default obsidian
docker exec basic-memory-server basic-memory project default obsidian
# Sync to index all files
docker exec basic-memory-server basic-memory sync
+31
View File
@@ -0,0 +1,31 @@
name: Daily Traction Report
on:
schedule:
- cron: '0 14 * * 1-5' # 9 AM EST weekdays
workflow_dispatch: # Manual trigger for testing
jobs:
generate-report:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.9'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r scripts/requirements.txt
- name: Generate Daily Traction Report
run: python scripts/daily_report.py
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
DISCORD_WEBHOOK: ${{ secrets.DISCORD_WEBHOOK }}
REDDIT_CLIENT_ID: ${{ secrets.REDDIT_CLIENT_ID }}
REDDIT_SECRET: ${{ secrets.REDDIT_SECRET }}
YOUTUBE_API_KEY: ${{ secrets.YOUTUBE_API_KEY }}
+373
View File
@@ -0,0 +1,373 @@
#!/usr/bin/env python3
"""
Basic Memory Daily Traction Report - Enhanced with Growth Tracking
Automated tracking across GitHub, Reddit, YouTube with daily change indicators
"""
import os
import requests
import json
from datetime import datetime, timedelta
import praw
from googleapiclient.discovery import build
from dateutil import parser
import base64
class BasicMemoryTracker:
def __init__(self):
self.github_token = os.getenv('GITHUB_TOKEN')
self.discord_webhook = os.getenv('DISCORD_WEBHOOK')
self.youtube_api_key = os.getenv('YOUTUBE_API_KEY')
# Reddit setup
self.reddit = praw.Reddit(
client_id=os.getenv('REDDIT_CLIENT_ID'),
client_secret=os.getenv('REDDIT_SECRET'),
user_agent='BasicMemoryTracker:v1.0'
)
# YouTube setup
self.youtube = build('youtube', 'v3', developerKey=self.youtube_api_key)
self.repo_owner = 'basicmachines-co'
self.repo_name = 'basic-memory'
self.youtube_channel = 'basicmachines-co'
self.metrics_file = 'data/daily_metrics.json'
def get_previous_metrics(self):
"""Get yesterday's metrics from GitHub repo storage"""
try:
headers = {'Authorization': f'token {self.github_token}'}
url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/contents/{self.metrics_file}'
response = requests.get(url, headers=headers)
if response.status_code == 200:
file_data = response.json()
content = base64.b64decode(file_data['content']).decode('utf-8')
return json.loads(content)
else:
print("📝 No previous metrics found - this is the first run!")
return {}
except Exception as e:
print(f"⚠️ Could not load previous metrics: {e}")
return {}
def save_current_metrics(self, metrics):
"""Save today's metrics to GitHub repo for tomorrow's comparison"""
try:
headers = {'Authorization': f'token {self.github_token}'}
# Prepare data
metrics_data = {
'date': datetime.now().isoformat(),
'metrics': metrics
}
content = json.dumps(metrics_data, indent=2)
encoded_content = base64.b64encode(content.encode()).decode()
# Check if file exists
file_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/contents/{self.metrics_file}'
existing_response = requests.get(file_url, headers=headers)
payload = {
'message': f'📊 Daily metrics update - {datetime.now().strftime("%Y-%m-%d")}',
'content': encoded_content
}
if existing_response.status_code == 200:
# File exists, update it
payload['sha'] = existing_response.json()['sha']
response = requests.put(file_url, headers=headers, json=payload)
else:
# File doesn't exist, create it
response = requests.put(file_url, headers=headers, json=payload)
if response.status_code in [200, 201]:
print("✅ Metrics saved for tomorrow's comparison!")
else:
print(f"⚠️ Failed to save metrics: {response.status_code}")
except Exception as e:
print(f"⚠️ Error saving metrics: {e}")
def calculate_change(self, current, previous, key):
"""Calculate the change between current and previous values"""
if not previous or key not in previous:
return 0, "🆕"
change = current - previous[key]
if change > 0:
return change, "📈"
elif change < 0:
return abs(change), "📉"
else:
return 0, "➡️"
def format_change(self, change, direction):
"""Format the change indicator for display"""
if direction == "🆕":
return "🆕"
elif direction == "📈":
return f"(+{change})"
elif direction == "📉":
return f"(-{change})"
else:
return "(±0)"
def get_github_metrics(self):
"""Get GitHub repository metrics"""
try:
headers = {'Authorization': f'token {self.github_token}'}
# Repository stats
repo_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}'
repo_response = requests.get(repo_url, headers=headers)
repo_data = repo_response.json()
# Traffic stats (requires push access)
traffic_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/traffic/views'
traffic_response = requests.get(traffic_url, headers=headers)
traffic_data = traffic_response.json() if traffic_response.status_code == 200 else {}
# Recent issues
issues_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/issues'
issues_response = requests.get(issues_url, headers=headers)
issues_data = issues_response.json() if issues_response.status_code == 200 else []
return {
'stars': repo_data.get('stargazers_count', 0),
'forks': repo_data.get('forks_count', 0),
'watchers': repo_data.get('watchers_count', 0),
'open_issues': repo_data.get('open_issues_count', 0),
'traffic_views': traffic_data.get('count', 0),
'traffic_unique': traffic_data.get('uniques', 0),
'recent_issues': len([i for i in issues_data if
parser.parse(i['created_at']).date() >= (datetime.now() - timedelta(days=1)).date()])
}
except Exception as e:
print(f"GitHub API error: {e}")
return {'error': str(e)}
def get_reddit_metrics(self):
"""Get Reddit metrics for Basic Memory mentions"""
try:
metrics = {
'total_mentions': 0,
'subreddit_members': 0,
'top_posts': [],
'hot_discussions': []
}
# Search for Basic Memory mentions
search_results = list(self.reddit.subreddit('all').search(
'Basic Memory', time_filter='day', limit=25
))
metrics['total_mentions'] = len(search_results)
# Get top posts
for post in search_results[:3]:
metrics['top_posts'].append({
'title': post.title[:50] + '...' if len(post.title) > 50 else post.title,
'score': post.score,
'subreddit': post.subreddit.display_name,
'num_comments': post.num_comments
})
# Check r/BasicMemory if it exists
try:
basic_memory_sub = self.reddit.subreddit('BasicMemory')
metrics['subreddit_members'] = basic_memory_sub.subscribers
except:
metrics['subreddit_members'] = 0
return metrics
except Exception as e:
print(f"Reddit API error: {e}")
return {'error': str(e)}
def get_youtube_metrics(self):
"""Get YouTube channel metrics"""
try:
# Get channel statistics
channel_response = self.youtube.channels().list(
part='statistics,snippet',
forUsername=self.youtube_channel
).execute()
if not channel_response['items']:
# Try by channel handle
search_response = self.youtube.search().list(
part='snippet',
q=f'@{self.youtube_channel}',
type='channel',
maxResults=1
).execute()
if search_response['items']:
channel_id = search_response['items'][0]['snippet']['channelId']
channel_response = self.youtube.channels().list(
part='statistics,snippet',
id=channel_id
).execute()
if channel_response['items']:
stats = channel_response['items'][0]['statistics']
return {
'subscribers': int(stats.get('subscriberCount', 0)),
'total_views': int(stats.get('viewCount', 0)),
'video_count': int(stats.get('videoCount', 0))
}
else:
return {'error': 'Channel not found'}
except Exception as e:
print(f"YouTube API error: {e}")
return {'error': str(e)}
def create_discord_embed(self, current_metrics, previous_metrics):
"""Create beautiful Discord embed with all metrics and growth indicators"""
github_data = current_metrics.get('github', {})
reddit_data = current_metrics.get('reddit', {})
youtube_data = current_metrics.get('youtube', {})
prev_github = previous_metrics.get('github', {})
prev_reddit = previous_metrics.get('reddit', {})
prev_youtube = previous_metrics.get('youtube', {})
# Calculate changes
star_change, star_dir = self.calculate_change(github_data.get('stars', 0), prev_github, 'stars')
sub_change, sub_dir = self.calculate_change(youtube_data.get('subscribers', 0), prev_youtube, 'subscribers')
view_change, view_dir = self.calculate_change(youtube_data.get('total_views', 0), prev_youtube, 'total_views')
reddit_change, reddit_dir = self.calculate_change(reddit_data.get('total_mentions', 0), prev_reddit, 'total_mentions')
member_change, member_dir = self.calculate_change(reddit_data.get('subreddit_members', 0), prev_reddit, 'subreddit_members')
# Calculate total reach
total_reach = (
github_data.get('traffic_unique', 0) +
reddit_data.get('total_mentions', 0) * 100 +
youtube_data.get('total_views', 0)
)
embed = {
"title": "🚀 Basic Memory Daily Traction Report",
"description": f"📅 {datetime.now().strftime('%A, %B %d, %Y')}",
"color": 0x00ff88,
"fields": [
{
"name": "⭐ GitHub Metrics",
"value": f"""
**Stars:** {github_data.get('stars', 'N/A')} {star_dir} {self.format_change(star_change, star_dir)}
**Forks:** {github_data.get('forks', 'N/A')} 🍴
**Traffic:** {github_data.get('traffic_unique', 'N/A')} unique visitors 👀
**Issues:** {github_data.get('recent_issues', 0)} new today 🐛
""".strip(),
"inline": True
},
{
"name": "🗨️ Reddit Activity",
"value": f"""
**Mentions:** {reddit_data.get('total_mentions', 'N/A')} {reddit_dir} {self.format_change(reddit_change, reddit_dir)}
**r/BasicMemory:** {reddit_data.get('subreddit_members', 'N/A')} {member_dir} {self.format_change(member_change, member_dir)}
**Hot Posts:** {len(reddit_data.get('top_posts', []))} trending 🔥
""".strip(),
"inline": True
},
{
"name": "📺 YouTube Stats",
"value": f"""
**Subscribers:** {youtube_data.get('subscribers', 'N/A')} {sub_dir} {self.format_change(sub_change, sub_dir)}
**Total Views:** {youtube_data.get('total_views', 'N/A'):,} {view_dir} {self.format_change(view_change, view_dir)}
**Videos:** {youtube_data.get('video_count', 'N/A')} 🎬
""".strip(),
"inline": True
}
],
"footer": {
"text": f"🤖 Automated by Basic Memory • Daily Reach: {total_reach:,}"
},
"timestamp": datetime.now().isoformat()
}
# Add top Reddit posts if available
if reddit_data.get('top_posts'):
top_post = reddit_data['top_posts'][0]
embed["fields"].append({
"name": "🔥 Top Reddit Post",
"value": f"**{top_post['title']}**\n📊 {top_post['score']} upvotes • 💬 {top_post['num_comments']} comments\n📍 r/{top_post['subreddit']}",
"inline": False
})
return embed
def send_discord_report(self, embed):
"""Send the report to Discord"""
try:
payload = {"embeds": [embed]}
response = requests.post(self.discord_webhook, json=payload)
if response.status_code == 204:
print("✅ Discord report sent successfully!")
return True
else:
print(f"❌ Discord webhook failed: {response.status_code}")
print(response.text)
return False
except Exception as e:
print(f"Discord send error: {e}")
return False
def run_daily_report(self):
"""Main function to generate and send daily report"""
print("🚀 Starting Basic Memory Daily Traction Report...")
# Load previous metrics
print("📊 Loading previous metrics...")
previous_metrics = self.get_previous_metrics()
# Collect current metrics
print("📊 Collecting GitHub metrics...")
github_data = self.get_github_metrics()
print("🗨️ Collecting Reddit metrics...")
reddit_data = self.get_reddit_metrics()
print("📺 Collecting YouTube metrics...")
youtube_data = self.get_youtube_metrics()
# Combine current metrics
current_metrics = {
'github': github_data,
'reddit': reddit_data,
'youtube': youtube_data
}
# Create and send report
print("🎨 Creating Discord embed with growth tracking...")
embed = self.create_discord_embed(current_metrics, previous_metrics.get('metrics', {}))
print("📤 Sending to Discord...")
success = self.send_discord_report(embed)
# Save current metrics for tomorrow
print("💾 Saving metrics for tomorrow's comparison...")
self.save_current_metrics(current_metrics)
if success:
print("🎉 Daily traction report completed successfully!")
else:
print("😞 Report failed to send")
# Print summary for GitHub Actions logs
print(f"""
📊 DAILY SUMMARY:
⭐ GitHub Stars: {github_data.get('stars', 'Error')}
👥 Reddit Mentions: {reddit_data.get('total_mentions', 'Error')}
📺 YouTube Subscribers: {youtube_data.get('subscribers', 'Error')}
📺 YouTube Views: {youtube_data.get('total_views', 'Error')}
""")
if __name__ == "__main__":
tracker = BasicMemoryTracker()
tracker.run_daily_report()
@@ -0,0 +1,4 @@
requests==2.31.0
praw==7.7.1
google-api-python-client==2.108.0
python-dateutil==2.8.2
+389
View File
@@ -0,0 +1,389 @@
#!/usr/bin/env python3
"""
Basic Memory Daily Traction Report - Enhanced with Growth Tracking
Automated tracking across GitHub, Reddit, YouTube with daily change indicators
"""
import os
import requests
import json
from datetime import datetime, timedelta
import praw
from googleapiclient.discovery import build
from dateutil import parser
import base64
class BasicMemoryTracker:
def __init__(self):
self.github_token = os.getenv('GITHUB_TOKEN')
self.discord_webhook = os.getenv('DISCORD_WEBHOOK')
self.youtube_api_key = os.getenv('YOUTUBE_API_KEY')
# Reddit setup
self.reddit = praw.Reddit(
client_id=os.getenv('REDDIT_CLIENT_ID'),
client_secret=os.getenv('REDDIT_SECRET'),
user_agent='BasicMemoryTracker:v1.0'
)
# YouTube setup
self.youtube = build('youtube', 'v3', developerKey=self.youtube_api_key)
self.repo_owner = 'basicmachines-co'
self.repo_name = 'basic-memory'
self.youtube_channel = 'basicmachines-co'
self.metrics_file = 'data/daily_metrics.json'
def get_previous_metrics(self):
"""Get yesterday's metrics from GitHub repo storage"""
try:
headers = {'Authorization': f'token {self.github_token}'}
url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/contents/{self.metrics_file}'
response = requests.get(url, headers=headers)
if response.status_code == 200:
file_data = response.json()
content = base64.b64decode(file_data['content']).decode('utf-8')
return json.loads(content)
else:
print("📝 No previous metrics found - this is the first run!")
return {}
except Exception as e:
print(f"⚠️ Could not load previous metrics: {e}")
return {}
def save_current_metrics(self, metrics):
"""Save today's metrics to GitHub repo for tomorrow's comparison"""
try:
headers = {'Authorization': f'token {self.github_token}'}
# Prepare data
metrics_data = {
'date': datetime.now().isoformat(),
'metrics': metrics
}
content = json.dumps(metrics_data, indent=2)
encoded_content = base64.b64encode(content.encode()).decode()
# Check if file exists
file_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/contents/{self.metrics_file}'
existing_response = requests.get(file_url, headers=headers)
payload = {
'message': f'📊 Daily metrics update - {datetime.now().strftime("%Y-%m-%d")}',
'content': encoded_content
}
if existing_response.status_code == 200:
# File exists, update it
payload['sha'] = existing_response.json()['sha']
response = requests.put(file_url, headers=headers, json=payload)
else:
# File doesn't exist, create it
response = requests.put(file_url, headers=headers, json=payload)
if response.status_code in [200, 201]:
print("✅ Metrics saved for tomorrow's comparison!")
else:
print(f"⚠️ Failed to save metrics: {response.status_code}")
except Exception as e:
print(f"⚠️ Error saving metrics: {e}")
def calculate_change(self, current, previous, key):
"""Calculate the change between current and previous values"""
if not previous or key not in previous:
return 0, "🆕"
change = current - previous[key]
if change > 0:
return change, "📈"
elif change < 0:
return abs(change), "📉"
else:
return 0, "➡️"
def format_change(self, change, direction):
"""Format the change indicator for display"""
if direction == "🆕":
return "🆕"
elif direction == "📈":
return f"(+{change})"
elif direction == "📉":
return f"(-{change})"
else:
return "(±0)"
def get_github_metrics(self):
"""Get GitHub repository metrics"""
try:
headers = {'Authorization': f'token {self.github_token}'}
# Repository stats
repo_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}'
repo_response = requests.get(repo_url, headers=headers)
repo_data = repo_response.json()
# Traffic stats (requires push access)
traffic_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/traffic/views'
traffic_response = requests.get(traffic_url, headers=headers)
traffic_data = traffic_response.json() if traffic_response.status_code == 200 else {}
# Recent issues
issues_url = f'https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/issues'
issues_response = requests.get(issues_url, headers=headers)
issues_data = issues_response.json() if issues_response.status_code == 200 else []
return {
'stars': repo_data.get('stargazers_count', 0),
'forks': repo_data.get('forks_count', 0),
'watchers': repo_data.get('watchers_count', 0),
'open_issues': repo_data.get('open_issues_count', 0),
'traffic_views': traffic_data.get('count', 0),
'traffic_unique': traffic_data.get('uniques', 0),
'recent_issues': len([i for i in issues_data if
parser.parse(i['created_at']).date() >= (datetime.now() - timedelta(days=1)).date()])
}
except Exception as e:
print(f"GitHub API error: {e}")
return {'error': str(e)}
def get_reddit_metrics(self):
"""Get Reddit metrics for Basic Memory mentions"""
try:
metrics = {
'total_mentions': 0,
'subreddit_members': 0,
'top_posts': [],
'hot_discussions': []
}
# Search for Basic Memory mentions
search_results = list(self.reddit.subreddit('all').search(
'Basic Memory', time_filter='day', limit=25
))
metrics['total_mentions'] = len(search_results)
# Get top posts
for post in search_results[:3]:
metrics['top_posts'].append({
'title': post.title[:50] + '...' if len(post.title) > 50 else post.title,
'score': post.score,
'subreddit': post.subreddit.display_name,
'num_comments': post.num_comments
})
# Check r/BasicMemory if it exists
try:
basic_memory_sub = self.reddit.subreddit('BasicMemory')
metrics['subreddit_members'] = basic_memory_sub.subscribers
except:
metrics['subreddit_members'] = 0
return metrics
except Exception as e:
print(f"Reddit API error: {e}")
return {'error': str(e)}
def get_youtube_metrics(self):
"""Get YouTube channel metrics"""
try:
# Try to find channel by handle first
search_response = self.youtube.search().list(
part='snippet',
q='basicmachines-co',
type='channel',
maxResults=5
).execute()
channel_id = None
if search_response.get('items'):
# Look for exact match or best match
for item in search_response['items']:
channel_title = item['snippet']['title'].lower()
if 'basic' in channel_title and 'machine' in channel_title:
channel_id = item['snippet']['channelId']
break
# If no exact match, use first result
if not channel_id and search_response['items']:
channel_id = search_response['items'][0]['snippet']['channelId']
if channel_id:
# Get channel statistics
channel_response = self.youtube.channels().list(
part='statistics,snippet',
id=channel_id
).execute()
if channel_response.get('items'):
stats = channel_response['items'][0]['statistics']
return {
'subscribers': int(stats.get('subscriberCount', 0)),
'total_views': int(stats.get('viewCount', 0)),
'video_count': int(stats.get('videoCount', 0))
}
# Fallback: return placeholder data
print("⚠️ YouTube channel not found, using placeholder data")
return {
'subscribers': 0,
'total_views': 0,
'video_count': 0
}
except Exception as e:
print(f"YouTube API error: {e}")
return {
'subscribers': 0,
'total_views': 0,
'video_count': 0
}
def create_discord_embed(self, current_metrics, previous_metrics):
"""Create beautiful Discord embed with all metrics and growth indicators"""
github_data = current_metrics.get('github', {})
reddit_data = current_metrics.get('reddit', {})
youtube_data = current_metrics.get('youtube', {})
prev_github = previous_metrics.get('github', {})
prev_reddit = previous_metrics.get('reddit', {})
prev_youtube = previous_metrics.get('youtube', {})
# Calculate changes
star_change, star_dir = self.calculate_change(github_data.get('stars', 0), prev_github, 'stars')
sub_change, sub_dir = self.calculate_change(youtube_data.get('subscribers', 0), prev_youtube, 'subscribers')
view_change, view_dir = self.calculate_change(youtube_data.get('total_views', 0), prev_youtube, 'total_views')
reddit_change, reddit_dir = self.calculate_change(reddit_data.get('total_mentions', 0), prev_reddit, 'total_mentions')
member_change, member_dir = self.calculate_change(reddit_data.get('subreddit_members', 0), prev_reddit, 'subreddit_members')
# Calculate total reach
total_reach = (
github_data.get('traffic_unique', 0) +
reddit_data.get('total_mentions', 0) * 100 +
youtube_data.get('total_views', 0)
)
embed = {
"title": "🚀 Basic Memory Daily Traction Report",
"description": f"📅 {datetime.now().strftime('%A, %B %d, %Y')}",
"color": 0x00ff88,
"fields": [
{
"name": "⭐ GitHub Metrics",
"value": f"""
**Stars:** {github_data.get('stars', 'N/A')} {star_dir} {self.format_change(star_change, star_dir)}
**Forks:** {github_data.get('forks', 'N/A')} 🍴
**Traffic:** {github_data.get('traffic_unique', 'N/A')} unique visitors 👀
**Issues:** {github_data.get('recent_issues', 0)} new today 🐛
""".strip(),
"inline": True
},
{
"name": "🗨️ Reddit Activity",
"value": f"""
**Mentions:** {reddit_data.get('total_mentions', 'N/A')} {reddit_dir} {self.format_change(reddit_change, reddit_dir)}
**r/BasicMemory:** {reddit_data.get('subreddit_members', 'N/A')} {member_dir} {self.format_change(member_change, member_dir)}
**Hot Posts:** {len(reddit_data.get('top_posts', []))} trending 🔥
""".strip(),
"inline": True
},
{
"name": "📺 YouTube Stats",
"value": f"""
**Subscribers:** {youtube_data.get('subscribers', 'N/A')} {sub_dir} {self.format_change(sub_change, sub_dir)}
**Total Views:** {youtube_data.get('total_views', 'N/A')} {view_dir} {self.format_change(view_change, view_dir)}
**Videos:** {youtube_data.get('video_count', 'N/A')} 🎬
""".strip(),
"inline": True
}
],
"footer": {
"text": f"🤖 Automated by Basic Memory • Daily Reach: {total_reach}"
},
"timestamp": datetime.now().isoformat()
}
# Add top Reddit posts if available
if reddit_data.get('top_posts'):
top_post = reddit_data['top_posts'][0]
embed["fields"].append({
"name": "🔥 Top Reddit Post",
"value": f"**{top_post['title']}**\n📊 {top_post['score']} upvotes • 💬 {top_post['num_comments']} comments\n📍 r/{top_post['subreddit']}",
"inline": False
})
return embed
def send_discord_report(self, embed):
"""Send the report to Discord"""
try:
payload = {"embeds": [embed]}
response = requests.post(self.discord_webhook, json=payload)
if response.status_code == 204:
print("✅ Discord report sent successfully!")
return True
else:
print(f"❌ Discord webhook failed: {response.status_code}")
print(response.text)
return False
except Exception as e:
print(f"Discord send error: {e}")
return False
def run_daily_report(self):
"""Main function to generate and send daily report"""
print("🚀 Starting Basic Memory Daily Traction Report...")
# Load previous metrics
print("📊 Loading previous metrics...")
previous_metrics = self.get_previous_metrics()
# Collect current metrics
print("📊 Collecting GitHub metrics...")
github_data = self.get_github_metrics()
print("🗨️ Collecting Reddit metrics...")
reddit_data = self.get_reddit_metrics()
print("📺 Collecting YouTube metrics...")
youtube_data = self.get_youtube_metrics()
# Combine current metrics
current_metrics = {
'github': github_data,
'reddit': reddit_data,
'youtube': youtube_data
}
# Create and send report
print("🎨 Creating Discord embed with growth tracking...")
embed = self.create_discord_embed(current_metrics, previous_metrics.get('metrics', {}))
print("📤 Sending to Discord...")
success = self.send_discord_report(embed)
# Save current metrics for tomorrow
print("💾 Saving metrics for tomorrow's comparison...")
self.save_current_metrics(current_metrics)
if success:
print("🎉 Daily traction report completed successfully!")
else:
print("😞 Report failed to send")
# Print summary for GitHub Actions logs
print(f"""
📊 DAILY SUMMARY:
⭐ GitHub Stars: {github_data.get('stars', 'Error')}
👥 Reddit Mentions: {reddit_data.get('total_mentions', 'Error')}
📺 YouTube Subscribers: {youtube_data.get('subscribers', 'Error')}
📺 YouTube Views: {youtube_data.get('total_views', 'Error')}
""")
if __name__ == "__main__":
tracker = BasicMemoryTracker()
tracker.run_daily_report()
+4
View File
@@ -0,0 +1,4 @@
requests==2.31.0
praw==7.7.1
google-api-python-client==2.108.0
python-dateutil==2.8.2
+4 -4
View File
@@ -20,24 +20,24 @@ from basic_memory.mcp.tools.edit_note import edit_note
from basic_memory.mcp.tools.move_note import move_note
from basic_memory.mcp.tools.sync_status import sync_status
from basic_memory.mcp.tools.project_management import (
list_projects,
list_memory_projects,
switch_project,
get_current_project,
set_default_project,
create_project,
create_memory_project,
delete_project,
)
__all__ = [
"build_context",
"canvas",
"create_project",
"create_memory_project",
"delete_note",
"delete_project",
"edit_note",
"get_current_project",
"list_directory",
"list_projects",
"list_memory_projects",
"move_note",
"read_content",
"read_note",
+2 -2
View File
@@ -109,7 +109,7 @@ def _format_cross_project_error_response(
```
## Available projects:
Use `list_projects()` to see all available projects and `switch_project("project-name")` to change projects.
Use `list_memory_projects()` to see all available projects and `switch_project("project-name")` to change projects.
""").strip()
@@ -153,7 +153,7 @@ def _format_potential_cross_project_guidance(
### To see all projects:
```
list_projects()
list_memory_projects()
```
""").strip()
@@ -19,7 +19,7 @@ from basic_memory.utils import generate_permalink
@mcp.tool("list_memory_projects")
async def list_projects(ctx: Context | None = None) -> str:
async def list_memory_projects(ctx: Context | None = None) -> str:
"""List all available projects with their status.
Shows all Basic Memory projects that are available, indicating which one
@@ -29,7 +29,7 @@ async def list_projects(ctx: Context | None = None) -> str:
Formatted list of projects with status indicators
Example:
list_projects()
list_memory_projects()
"""
if ctx: # pragma: no cover
await ctx.info("Listing all available projects")
@@ -150,7 +150,7 @@ async def switch_project(project_name: str, ctx: Context | None = None) -> str:
4. **Try again**: The error might be temporary
## Available options:
- See all projects: `list_projects()`
- See all projects: `list_memory_projects()`
- Stay on current project: `get_current_project()`
- Try different project: `switch_project("correct-project-name")`
@@ -231,7 +231,7 @@ async def set_default_project(project_name: str, ctx: Context | None = None) ->
@mcp.tool("create_memory_project")
async def create_project(
async def create_memory_project(
project_name: str, project_path: str, set_default: bool = False, ctx: Context | None = None
) -> str:
"""Create a new Basic Memory project.
@@ -248,8 +248,8 @@ async def create_project(
Confirmation message with project details
Example:
create_project("my-research", "~/Documents/research")
create_project("work-notes", "/home/user/work", set_default=True)
create_memory_project("my-research", "~/Documents/research")
create_memory_project("work-notes", "/home/user/work", set_default=True)
"""
if ctx: # pragma: no cover
await ctx.info(f"Creating project: {project_name} at {project_path}")
+4 -4
View File
@@ -67,13 +67,13 @@ def _format_search_error_response(error_message: str, query: str, search_type: s
The current project is not accessible or doesn't exist: {error_message}
## How to resolve:
1. **Check available projects**: `list_projects()`
1. **Check available projects**: `list_memory_projects()`
2. **Switch to valid project**: `switch_project("valid-project-name")`
3. **Verify project setup**: Ensure your project is properly configured
## Current session info:
- Check current project: `get_current_project()`
- See available projects: `list_projects()`
- See available projects: `list_memory_projects()`
""").strip()
# No results found
@@ -162,7 +162,7 @@ You don't have permission to search in the current project: {error_message}
3. **Check authentication**: You might need to re-authenticate
## Alternative actions:
- List available projects: `list_projects()`
- List available projects: `list_memory_projects()`
- Switch to accessible project: `switch_project("project-name")`
- Check current project: `get_current_project()`"""
@@ -189,7 +189,7 @@ Error searching for '{query}': {error_message}
- **Browse files**: `list_directory("/")` - See all available content
- **Recent activity**: `recent_activity(timeframe="7d")` - Check what's been updated
- **Project info**: `get_current_project()` - Verify current project
- **All projects**: `list_projects()` - Switch to different project if needed
- **All projects**: `list_memory_projects()` - Switch to different project if needed
## Search syntax reference:
- **Basic**: `keyword` or `multiple words`
+1 -1
View File
@@ -594,7 +594,7 @@ async def test_move_note_potential_cross_project_guidance(mcp_server, app):
error_message = move_result[0].text
assert "Check Project Context" in error_message
assert "workspace-docs" in error_message # Should mention other available projects
assert "list_projects" in error_message
assert "list_memory_projects" in error_message
assert "switch_project" in error_message
+1 -1
View File
@@ -93,7 +93,7 @@ class TestMCPServer:
# Missing SUPABASE_ANON_KEY
}
with patch.dict(os.environ, env_vars):
with patch.dict(os.environ, env_vars, clear=True):
with pytest.raises(ValueError, match="SUPABASE_URL and SUPABASE_ANON_KEY must be set"):
create_auth_config()