"""
connectd - groq message drafting
reads soul from file, uses as guideline for llm to personalize
"""
import os
import json
from groq import Groq
GROQ_API_KEY = os.getenv("GROQ_API_KEY")
GROQ_MODEL = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile")
client = Groq(api_key=GROQ_API_KEY) if GROQ_API_KEY else None
# load soul from file (guideline, not script)
SOUL_PATH = os.getenv("SOUL_PATH", "/app/soul.txt")
def load_soul():
try:
with open(SOUL_PATH, 'r') as f:
return f.read().strip()
except:
return None
SIGNATURE_HTML = """
"""
SIGNATURE_PLAINTEXT = """
---
github.com/sudoxnym/connectd (main repo)
github: github.com/connectd-daemon
mastodon: @connectd@mastodon.sudoxreboot.com
bluesky: connectd.bsky.social
lemmy: lemmy.sudoxreboot.com/c/connectd
discord: discord.gg/connectd
matrix: @connectd:sudoxreboot.com
reddit: reddit.com/r/connectd
email: connectd@sudoxreboot.com
"""
def draft_intro_with_llm(match_data: dict, recipient: str = 'a', dry_run: bool = True):
"""
draft an intro message using groq llm.
args:
match_data: dict with human_a, human_b, overlap_score, overlap_reasons
recipient: 'a' or 'b' - who receives the message
dry_run: if True, preview mode
returns:
tuple (result_dict, error_string)
result_dict has: subject, draft_html, draft_plain
"""
if not client:
return None, "GROQ_API_KEY not set"
try:
human_a = match_data.get('human_a', {})
human_b = match_data.get('human_b', {})
reasons = match_data.get('overlap_reasons', [])
# recipient gets the message, about_person is who we're introducing them to
if recipient == 'a':
to_person = human_a
about_person = human_b
else:
to_person = human_b
about_person = human_a
to_name = to_person.get('username', 'friend')
about_name = about_person.get('username', 'someone')
about_bio = about_person.get('extra', {}).get('bio', '')
# extract contact info for about_person
about_extra = about_person.get('extra', {})
if isinstance(about_extra, str):
import json as _json
about_extra = _json.loads(about_extra) if about_extra else {}
about_contact = about_person.get('contact', {})
if isinstance(about_contact, str):
about_contact = _json.loads(about_contact) if about_contact else {}
# build contact link for about_person
about_platform = about_person.get('platform', '')
about_username = about_person.get('username', '')
contact_link = None
if about_platform == 'mastodon' and about_username:
if '@' in about_username:
parts = about_username.split('@')
if len(parts) >= 2:
contact_link = f"https://{parts[1]}/@{parts[0]}"
elif about_platform == 'github' and about_username:
contact_link = f"https://github.com/{about_username}"
elif about_extra.get('mastodon') or about_contact.get('mastodon'):
handle = about_extra.get('mastodon') or about_contact.get('mastodon')
if '@' in handle:
parts = handle.lstrip('@').split('@')
if len(parts) >= 2:
contact_link = f"https://{parts[1]}/@{parts[0]}"
elif about_extra.get('github') or about_contact.get('github'):
contact_link = f"https://github.com/{about_extra.get('github') or about_contact.get('github')}"
elif about_extra.get('email'):
contact_link = about_extra['email']
elif about_contact.get('email'):
contact_link = about_contact['email']
elif about_extra.get('website'):
contact_link = about_extra['website']
elif about_extra.get('external_links', {}).get('website'):
contact_link = about_extra['external_links']['website']
elif about_extra.get('extra', {}).get('website'):
contact_link = about_extra['extra']['website']
elif about_platform == 'reddit' and about_username:
contact_link = f"reddit.com/u/{about_username}"
if not contact_link:
contact_link = f"github.com/{about_username}" if about_username else "reach out via connectd"
# skip if no real contact method (just reddit or generic)
if contact_link.startswith('reddit.com') or contact_link == "reach out via connectd" or 'stackblitz' in contact_link:
return None, f"no real contact info for {about_name} - skipping draft"
# format the shared factors naturally
if reasons:
factor = ', '.join(reasons[:3]) if len(reasons) > 1 else reasons[0]
else:
factor = "shared values and interests"
# load soul as guideline
soul = load_soul()
if not soul:
return None, "could not load soul file"
# build the prompt - soul is GUIDELINE not script
prompt = f"""you are connectd, a daemon that finds isolated builders and connects them.
write a personal message TO {to_name} telling them about {about_name}.
here is the soul/spirit of what connectd is about - use this as a GUIDELINE for tone and message, NOT as a script to copy verbatim:
---
{soul}
---
key facts for this message:
- recipient: {to_name}
- introducing them to: {about_name}
- their shared interests/values: {factor}
- about {about_name}: {about_bio if about_bio else 'a builder like you'}
- HOW TO REACH {about_name}: {contact_link}
RULES:
1. say their name ONCE at start, then use "you"
2. MUST include how to reach {about_name}: {contact_link}
3. lowercase, raw, emotional - follow the soul
4. end with the contact link
return ONLY the message body. signature is added separately."""
response = client.chat.completions.create(
model=GROQ_MODEL,
messages=[{"role": "user", "content": prompt}],
temperature=0.6,
max_tokens=1200
)
body = response.choices[0].message.content.strip()
# generate subject
subject_prompt = f"""generate a short, lowercase email subject for a message to {to_name} about connecting them with {about_name} over their shared interest in {factor}.
no corporate speak. no clickbait. raw and real.
examples:
- "found you, {to_name}"
- "you're not alone"
- "a door just opened"
- "{to_name}, there's someone you should meet"
return ONLY the subject line."""
subject_response = client.chat.completions.create(
model=GROQ_MODEL,
messages=[{"role": "user", "content": subject_prompt}],
temperature=0.9,
max_tokens=50
)
subject = subject_response.choices[0].message.content.strip().strip('"').strip("'")
# format html
draft_html = f"{body}
{SIGNATURE_HTML}"
draft_plain = body + SIGNATURE_PLAINTEXT
return {
'subject': subject,
'draft_html': draft_html,
'draft_plain': draft_plain
}, None
except Exception as e:
return None, str(e)
# for backwards compat with old code
def draft_message(person: dict, factor: str, platform: str = "email") -> dict:
"""legacy function - wraps new api"""
match_data = {
'human_a': {'username': 'recipient'},
'human_b': person,
'overlap_reasons': [factor]
}
result, error = draft_intro_with_llm(match_data, recipient='a')
if error:
raise ValueError(error)
return {
'subject': result['subject'],
'body_html': result['draft_html'],
'body_plain': result['draft_plain']
}
if __name__ == "__main__":
# test
test_data = {
'human_a': {'username': 'sudoxnym', 'extra': {'bio': 'building intentional communities'}},
'human_b': {'username': 'testuser', 'extra': {'bio': 'home assistant enthusiast'}},
'overlap_reasons': ['home-assistant', 'open source', 'community building']
}
result, error = draft_intro_with_llm(test_data, recipient='a')
if error:
print(f"error: {error}")
else:
print(f"subject: {result['subject']}")
print(f"\nbody:\n{result['draft_plain']}")
# contact method ranking - USAGE BASED
# we rank by where the person is MOST ACTIVE, not by our preference
def determine_contact_method(human):
"""
determine ALL available contact methods, ranked by USER'S ACTIVITY.
looks at activity metrics to decide where they're most engaged.
returns: (best_method, best_info, fallbacks)
where fallbacks is a list of (method, info) tuples in activity order
"""
import json
extra = human.get('extra', {})
contact = human.get('contact', {})
if isinstance(extra, str):
extra = json.loads(extra) if extra else {}
if isinstance(contact, str):
contact = json.loads(contact) if contact else {}
nested_extra = extra.get('extra', {})
platform = human.get('platform', '')
available = []
# === ACTIVITY SCORING ===
# each method gets scored by how active the user is there
# EMAIL - always medium priority (we cant measure activity)
email = extra.get('email') or contact.get('email') or nested_extra.get('email')
if email and '@' in str(email):
available.append(('email', email, 50)) # baseline score
# MASTODON - score by post count / followers
mastodon = extra.get('mastodon') or contact.get('mastodon') or nested_extra.get('mastodon')
if mastodon:
masto_activity = extra.get('mastodon_posts', 0) or extra.get('statuses_count', 0)
masto_score = min(100, 30 + (masto_activity // 10)) # 30 base + 1 per 10 posts
available.append(('mastodon', mastodon, masto_score))
# if they CAME FROM mastodon, thats their primary
if platform == 'mastodon':
handle = f"@{human.get('username')}"
instance = human.get('instance') or extra.get('instance') or ''
if instance:
handle = f"@{human.get('username')}@{instance}"
activity = extra.get('statuses_count', 0) or extra.get('activity_count', 0)
score = min(100, 50 + (activity // 5)) # higher base since its their home
# dont dupe
if not any(a[0] == 'mastodon' for a in available):
available.append(('mastodon', handle, score))
else:
# update score if this is higher
for i, (m, info, s) in enumerate(available):
if m == 'mastodon' and score > s:
available[i] = ('mastodon', handle, score)
# MATRIX - score by presence (binary for now)
matrix = extra.get('matrix') or contact.get('matrix') or nested_extra.get('matrix')
if matrix and ':' in str(matrix):
available.append(('matrix', matrix, 40))
# BLUESKY - score by followers/posts if available
bluesky = extra.get('bluesky') or contact.get('bluesky') or nested_extra.get('bluesky')
if bluesky:
bsky_activity = extra.get('bluesky_posts', 0)
bsky_score = min(100, 25 + (bsky_activity // 10))
available.append(('bluesky', bluesky, bsky_score))
# LEMMY - score by activity
lemmy = extra.get('lemmy') or contact.get('lemmy') or nested_extra.get('lemmy')
if lemmy:
lemmy_activity = extra.get('lemmy_posts', 0) or extra.get('lemmy_comments', 0)
lemmy_score = min(100, 30 + lemmy_activity)
available.append(('lemmy', lemmy, lemmy_score))
if platform == 'lemmy':
handle = human.get('username')
activity = extra.get('activity_count', 0)
score = min(100, 50 + activity)
if not any(a[0] == 'lemmy' for a in available):
available.append(('lemmy', handle, score))
# DISCORD - lower priority (hard to DM)
discord = extra.get('discord') or contact.get('discord') or nested_extra.get('discord')
if discord:
available.append(('discord', discord, 20))
# GITHUB ISSUE - for github users, score by repo activity
if platform == 'github':
top_repos = extra.get('top_repos', [])
if top_repos:
repo = top_repos[0] if isinstance(top_repos[0], str) else top_repos[0].get('name', '')
stars = extra.get('total_stars', 0)
repos_count = extra.get('repos_count', 0)
# active github user = higher issue score
gh_score = min(60, 20 + (stars // 100) + (repos_count // 5))
if repo:
available.append(('github_issue', f"{human.get('username')}/{repo}", gh_score))
# FORGE ISSUE - for self-hosted git users (gitea/forgejo/gitlab/sourcehut/codeberg)
# these are HIGH SIGNAL users - they actually selfhost
if platform and ':' in platform:
platform_type, instance = platform.split(':', 1)
if platform_type in ('gitea', 'forgejo', 'gogs', 'gitlab', 'sourcehut'):
repos = extra.get('repos', [])
if repos:
repo = repos[0] if isinstance(repos[0], str) else repos[0].get('name', '')
instance_url = extra.get('instance_url', '')
if repo and instance_url:
# forge users get higher priority than github (they selfhost!)
forge_score = 55 # higher than github_issue (50)
available.append(('forge_issue', {
'platform_type': platform_type,
'instance': instance,
'instance_url': instance_url,
'owner': human.get('username'),
'repo': repo
}, forge_score))
# REDDIT - discovered people, use their other links
if platform == 'reddit':
reddit_activity = extra.get('reddit_activity', 0) or extra.get('activity_count', 0)
# reddit users we reach via their external links (email, mastodon, etc)
# boost their other methods if reddit is their main platform
for i, (m, info, score) in enumerate(available):
if m in ('email', 'mastodon', 'matrix', 'bluesky'):
# boost score for reddit-discovered users' external contacts
boost = min(30, reddit_activity // 3)
available[i] = (m, info, score + boost)
# sort by activity score (highest first)
available.sort(key=lambda x: x[2], reverse=True)
if not available:
return 'manual', None, []
best = available[0]
fallbacks = [(m, i) for m, i, p in available[1:]]
return best[0], best[1], fallbacks
def get_ranked_contact_methods(human):
"""
get all contact methods for a human, ranked by their activity.
"""
method, info, fallbacks = determine_contact_method(human)
if method == 'manual':
return []
return [(method, info)] + fallbacks