mra/carbon-copy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

minor bug fixes

Browse source 
Co-authored-by: Peter Bierma <zintensitydev@gmail.com>
This commit is contained in:
Micha Albert 2025-07-30 11:40:41 -04:00
parent a36f6d40e3
commit 251e7f0a7c
No known key found for this signature in database
17 changed files with 178 additions and 1919 deletions
Download patch file Download diff file Expand all files Collapse all files

93
alembic.ini
View file

@ -1,93 +0,0 @@
[alembic]
# path to migration scripts
script_location = alembic
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version number format for new migration files
# version_num_format = YYYYMMDD_HHMMSS
# version name template to use
# version_path_separator = os.pathsep
sqlalchemy.url = sqlite:///./carbon_copy.db
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
# hooks = ruff
# ruff.type = exec
# ruff.executable = %(here)s/.venv/bin/ruff
# ruff.options = --fix REVISION_SCRIPT_FILENAME
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

107
alembic/env.py
View file

@ -1,107 +0,0 @@
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
import os
import sys
from pathlib import Path
# Add the project root to the Python path
project_root = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(project_root / "src"))
# Import your models
from carbon_copy.models import SQLModel
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
target_metadata = SQLModel.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def get_database_url():
"""Get database URL from environment or config."""
from dotenv import load_dotenv
load_dotenv()
db_url = os.getenv("DATABASE_URL")
if db_url:
return db_url
config_url = config.get_main_option("sqlalchemy.url")
if config_url:
return config_url
# Default fallback
return "sqlite:///./carbon_copy.db"
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = get_database_url()
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
configuration = config.get_section(config.config_ini_section) or {}
configuration["sqlalchemy.url"] = get_database_url()
connectable = engine_from_config(
configuration,
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection, target_metadata=target_metadata
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

24
alembic/script.py.mako
View file

@ -1,24 +0,0 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

43
init_db.py
View file

@ -1,43 +0,0 @@
#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
"""
Carbon Copy Database Initialization Script
This script initializes the database and loads seed data for the Carbon Copy
email-based multiplayer text adventure.
"""
import sys
from pathlib import Path
# Add the src directory to Python path
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root / "src"))
from carbon_copy.database import create_db_and_tables
from carbon_copy.seed_data import initialize_game_world
def main():
"""Initialize the database and load seed data."""
print("Carbon Copy - Database Initialization")
print("=" * 40)
print("Creating database tables...")
create_db_and_tables()
print("✓ Database tables created successfully")
print("\nLoading seed data...")
initialize_game_world()
print("✓ Seed data loaded successfully")
print("\nDatabase initialization complete!")
print("You can now start the FastAPI server with:")
print("uvicorn carbon_copy.main:app --reload")
if __name__ == "__main__":
main()

16
pyproject.toml
View file

@ -12,6 +12,7 @@ license = "MIT"
keywords = []
authors = [
{ name = "Micha Albert", email = "info@micha.zone" },
{ name = "Peter Bierma", email = "zintensitydev@gmail.com" }
]
classifiers = [
"Development Status :: 4 - Beta",
@ -21,20 +22,7 @@ classifiers = [
"Programming Language :: Python :: Implementation :: CPython",
]
dependencies = [
"fastapi~=0.116.1",
"uvicorn[standard]~=0.35.0",
"jinja2~=3.1.6",
"sqlmodel~=0.0.24",
"alembic~=1.16.4",
"asyncpg~=0.30.0",
"aiosqlite~=0.21.0",
"python-dotenv~=1.1.1",
"aiosmtplib~=4.0.1",
"aioimaplib~=2.0.1",
"email-validator~=2.2.0",
"passlib[bcrypt]~=1.7.4",
"python-jose[cryptography]~=3.5.0",
"python-multipart~=0.0.20",
"httpx~=0.28.1"
]
[project.urls]

4
src/carbon_copy/__about__.py
View file

@ -1,4 +0,0 @@
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
__version__ = "0.0.1"

6
src/carbon_copy/__init__.py
View file

@ -1,7 +1,3 @@
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
from .main import app
__all__ = ["app"]
# SPDX-License-Identifier: MIT

5
src/carbon_copy/__main__.py Normal file
View file

@ -0,0 +1,5 @@
def main():
pass
if __name__ == '__main__':
main()

81
src/carbon_copy/database.py
View file

@ -1,46 +1,57 @@
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
from __future__ import annotations
import os
from typing import AsyncGenerator
from sqlmodel import SQLModel, create_engine, Session
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from dotenv import load_dotenv
from abc import ABC, abstractmethod
from typing import NewType
from dataclasses import dataclass
import aiofiles
load_dotenv()
UUID = NewType("UUID", str)
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./carbon_copy.db")
class Cursor(ABC):
def __init__(self, database: Database, /) -> None:
self.database = database
# Convert SQLite URL to async version if needed
if DATABASE_URL.startswith("sqlite:///"):
ASYNC_DATABASE_URL = DATABASE_URL.replace("sqlite:///", "sqlite+aiosqlite:///")
elif DATABASE_URL.startswith("postgresql://"):
ASYNC_DATABASE_URL = DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://")
else:
ASYNC_DATABASE_URL = DATABASE_URL
@abstractmethod
async def write(self) -> None:
...
# Create engines
if "sqlite" in DATABASE_URL:
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
async_engine = create_async_engine(ASYNC_DATABASE_URL, connect_args={"check_same_thread": False})
else:
engine = create_engine(DATABASE_URL)
async_engine = create_async_engine(ASYNC_DATABASE_URL)
@abstractmethod
async def read(self) -> dict[str, str]:
pass
class Database(ABC):
def __init__(self) -> None:
...
def create_db_and_tables():
"""Create database tables."""
SQLModel.metadata.create_all(engine)
@abstractmethod
async def cursor(self, field_name: str, /) -> Cursor:
...
@abstractmethod
async def connect(self) -> None:
...
def get_session():
"""Get database session."""
with Session(engine) as session:
yield session
@abstractmethod
async def close(self) -> None:
...
@dataclass(slots=True)
class DatabaseItem(ABC):
database: Database
name: str
unique_id: UUID
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
"""Get async database session."""
async with AsyncSession(async_engine) as session:
yield session
@classmethod
@abstractmethod
async def find(cls, database: Database, name: str, /) -> Self | None:
async with database.cursor() as cursor:
return await cursor.find('rooms', name)
@classmethod
@abstractmethod
async def new(cls, database: Database, name: str, /) -> Self:
...
@abstractmethod
async def save(self) -> None:
...

483
src/carbon_copy/email_client.py
View file

@ -1,483 +0,0 @@
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
import os
import email as email_module
from typing import List, Optional, Dict, Any
from datetime import datetime, timezone
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import aiosmtplib
import aioimaplib
from dotenv import load_dotenv
load_dotenv()
class EmailConfig:
"""Email configuration from environment variables."""
def __init__(self):
# SMTP Configuration - all required
self.smtp_host = os.getenv("SMTP_HOST")
smtp_port_str = os.getenv("SMTP_PORT")
self.smtp_username = os.getenv("SMTP_USERNAME")
self.smtp_password = os.getenv("SMTP_PASSWORD")
self.smtp_use_tls = os.getenv("SMTP_USE_TLS", "true").lower() == "true"
# IMAP Configuration - all required
self.imap_host = os.getenv("IMAP_HOST")
imap_port_str = os.getenv("IMAP_PORT")
self.imap_username = os.getenv("IMAP_USERNAME")
self.imap_password = os.getenv("IMAP_PASSWORD")
self.imap_use_ssl = os.getenv("IMAP_USE_SSL", "true").lower() == "true"
# Game Configuration - required
self.game_master_email = os.getenv("GAME_MASTER_EMAIL")
# Validate required fields
required_fields = {
"SMTP_HOST": self.smtp_host,
"SMTP_PORT": smtp_port_str,
"SMTP_USERNAME": self.smtp_username,
"SMTP_PASSWORD": self.smtp_password,
"IMAP_HOST": self.imap_host,
"IMAP_PORT": imap_port_str,
"IMAP_USERNAME": self.imap_username,
"IMAP_PASSWORD": self.imap_password,
"GAME_MASTER_EMAIL": self.game_master_email,
}
missing_fields = [field for field, value in required_fields.items() if not value]
if missing_fields:
raise ValueError(f"Missing required environment variables: {', '.join(missing_fields)}")
if not smtp_port_str:
raise ValueError("SMTP_PORT must be set in the environment variables")
self.smtp_port = int(smtp_port_str)
if not imap_port_str:
raise ValueError("IMAP_PORT must be set in the environment variables")
self.imap_port = int(imap_port_str)
class EmailMessage:
"""Represents an email message."""
def __init__(
self,
subject: str,
content: str,
from_email: str,
to_email: str,
message_id: Optional[str] = None,
in_reply_to: Optional[str] = None,
thread_id: Optional[str] = None,
received_at: Optional[datetime] = None
):
self.subject = subject
self.content = content
self.from_email = from_email
self.to_email = to_email
self.message_id = message_id
self.in_reply_to = in_reply_to
self.thread_id = thread_id
self.received_at = received_at or datetime.now(timezone.utc)
class SMTPClient:
"""Async SMTP client for sending emails."""
def __init__(self, config: EmailConfig):
self.config = config
async def send_email(
self,
to_email: str,
subject: str,
content: str,
from_email: Optional[str] = None,
reply_to: Optional[str] = None,
message_id: Optional[str] = None,
is_html: bool = False
) -> bool:
"""Send an email via SMTP."""
try:
from_email = from_email or self.config.smtp_username
# Create message
if is_html:
msg = MIMEMultipart("alternative")
msg.attach(MIMEText(content, "html"))
else:
msg = MIMEText(content, "plain")
if not from_email:
raise ValueError("From email must be provided or set in configuration")
msg["Subject"] = subject
msg["From"] = from_email
msg["To"] = to_email
if reply_to:
msg["Reply-To"] = reply_to
msg["In-Reply-To"] = reply_to
msg["References"] = reply_to
if message_id:
msg["Message-ID"] = message_id
# Send email
async with aiosmtplib.SMTP(
hostname=self.config.smtp_host,
port=self.config.smtp_port,
use_tls=self.config.smtp_use_tls
) as smtp:
if not self.config.smtp_username or not self.config.smtp_password:
raise ValueError("SMTP username and password must be set in configuration")
await smtp.login(self.config.smtp_username, self.config.smtp_password)
await smtp.send_message(msg)
return True
except Exception as e:
print(f"Failed to send email: {e}")
return False
async def send_game_notification(
self,
to_email: str,
player_name: str,
notification_type: str,
content: str,
room_name: Optional[str] = None
) -> bool:
"""Send a game-specific notification email."""
subject_map = {
"welcome": f"Welcome to Carbon Copy, {player_name}!",
"room_update": f"Room Update: {room_name}",
"message": f"New Message in Carbon Copy",
"combat": f"Combat Update for {player_name}",
"item": f"Item Update for {player_name}",
"system": f"System Notification - Carbon Copy"
}
subject = subject_map.get(notification_type, "Carbon Copy Notification")
# Create HTML email content
html_content = f"""
<html>
<head>
<style>
body {{ font-family: Arial, sans-serif; }}
.header {{ background-color: #2c3e50; color: white; padding: 20px; }}
.content {{ padding: 20px; }}
.footer {{ background-color: #ecf0f1; padding: 10px; font-size: 12px; }}
</style>
</head>
<body>
<div class="header">
<h1>Carbon Copy</h1>
<p>Multiplayer Email Text Adventure</p>
</div>
<div class="content">
<h2>Hello, {player_name}!</h2>
{content}
<p><em>Reply to this email to send a message in the game.</em></p>
</div>
<div class="footer">
<p>This is an automated message from Carbon Copy.
To unsubscribe, reply with "UNSUBSCRIBE".</p>
</div>
</body>
</html>
"""
return await self.send_email(
to_email=to_email,
subject=subject,
content=html_content,
is_html=True
)
class IMAPClient:
"""Async IMAP client for receiving emails."""
def __init__(self, config: EmailConfig):
self.config = config
self.client: Optional[aioimaplib.IMAP4_SSL] = None
async def connect(self) -> bool:
"""Connect to IMAP server."""
try:
# Validate configuration
if not self.config.imap_host:
raise ValueError("IMAP host is not configured")
if not self.config.imap_username:
raise ValueError("IMAP username is not configured")
if not self.config.imap_password:
raise ValueError("IMAP password is not configured")
print(f"Connecting to IMAP server {self.config.imap_host}:{self.config.imap_port}")
self.client = aioimaplib.IMAP4_SSL(
host=self.config.imap_host,
port=self.config.imap_port
)
print("Waiting for server hello...")
await self.client.wait_hello_from_server()
print(f"Logging in as {self.config.imap_username}...")
await self.client.login(self.config.imap_username, self.config.imap_password)
print("Selecting INBOX...")
await self.client.select("INBOX")
print("Successfully connected to IMAP server")
return True
except Exception as e:
print(f"Failed to connect to IMAP: {e}")
print(f"IMAP Host: {self.config.imap_host}")
print(f"IMAP Port: {self.config.imap_port}")
print(f"IMAP Username: {self.config.imap_username}")
return False
async def disconnect(self):
"""Disconnect from IMAP server."""
if self.client:
await self.client.logout()
self.client = None
async def test_connection(self) -> bool:
"""Test IMAP connection and return detailed status."""
try:
print("Testing IMAP connection...")
success = await self.connect()
if success:
print("✓ IMAP connection test successful")
await self.disconnect()
return True
else:
print("✗ IMAP connection test failed")
return False
except Exception as e:
print(f"✗ IMAP connection test failed with exception: {e}")
return False
async def fetch_new_messages(self, mark_as_read: bool = False) -> List[EmailMessage]:
"""Fetch new (unread) messages from inbox."""
if not self.client:
print("No IMAP client connection, attempting to connect...")
if not await self.connect():
raise ConnectionError("Failed to connect to IMAP server")
if not self.client: # Type guard
raise ConnectionError("IMAP client is None after connection attempt")
try:
# Search for unseen messages
search_criteria = "UNSEEN"
print(f"Searching for messages with criteria: {search_criteria}")
response = await self.client.search(search_criteria)
if response.result != "OK":
raise RuntimeError(f"IMAP search failed with result: {response.result}")
# Handle message IDs - they come as bytes and need to be decoded
raw_message_ids = response.lines[0].split() if response.lines[0] else []
message_ids = []
for raw_id in raw_message_ids:
if isinstance(raw_id, bytes):
message_ids.append(raw_id.decode('utf-8'))
else:
message_ids.append(str(raw_id))
print(f"Found {len(message_ids)} unread messages")
messages = []
for msg_id in message_ids:
# Fetch message
print(f"Fetching message ID: {msg_id}")
fetch_response = await self.client.fetch(msg_id, "RFC822")
if fetch_response.result != "OK":
print(f"Failed to fetch message {msg_id}: {fetch_response.result}")
if hasattr(fetch_response, 'lines') and fetch_response.lines:
print(f"Server response: {fetch_response.lines}")
continue
# Parse message
raw_email = fetch_response.lines[1]
email_msg = email_module.message_from_bytes(raw_email)
# Extract message details
subject = email_msg.get("Subject", "")
from_email = email_msg.get("From", "")
to_email = email_msg.get("To", "")
message_id = email_msg.get("Message-ID", "")
in_reply_to = email_msg.get("In-Reply-To", "")
date_str = email_msg.get("Date", "")
# Parse date
received_at = None
if date_str:
try:
import email.utils
date_tuple = email.utils.parsedate_tz(date_str)
if date_tuple:
timestamp = email.utils.mktime_tz(date_tuple)
received_at = datetime.fromtimestamp(timestamp, tz=timezone.utc)
except:
pass
# Extract content
content = ""
if email_msg.is_multipart():
for part in email_msg.walk():
if part.get_content_type() == "text/plain":
payload = part.get_payload(decode=True)
if isinstance(payload, bytes):
content = payload.decode("utf-8", errors="ignore")
break
elif part.get_content_type() == "text/html" and not content:
payload = part.get_payload(decode=True)
if isinstance(payload, bytes):
content = payload.decode("utf-8", errors="ignore")
else:
payload = email_msg.get_payload(decode=True)
if isinstance(payload, bytes):
content = payload.decode("utf-8", errors="ignore")
elif isinstance(payload, str):
content = payload
# Create EmailMessage object
msg = EmailMessage(
subject=subject,
content=content,
from_email=from_email,
to_email=to_email,
message_id=message_id,
in_reply_to=in_reply_to,
received_at=received_at
)
messages.append(msg)
# Mark as read if requested
if mark_as_read:
await self.client.store(msg_id, "+FLAGS", "\\Seen")
return messages
except Exception as e:
print(f"Failed to fetch messages: {e}")
raise RuntimeError(f"Failed to fetch messages: {e}") from e
async def mark_as_read(self, message_id: str) -> bool:
"""Mark a specific message as read."""
if not self.client:
print("No IMAP client connection, attempting to connect...")
if not await self.connect():
raise ConnectionError("Failed to connect to IMAP server")
if not self.client: # Type guard
raise ConnectionError("IMAP client is None after connection attempt")
try:
# Search for the message by Message-ID
print(f"Searching for message with ID: {message_id}")
search_response = await self.client.search(f'HEADER Message-ID "{message_id}"')
if search_response.result != "OK":
raise RuntimeError(f"IMAP search failed with result: {search_response.result}")
if not search_response.lines[0]:
print(f"No message found with Message-ID: {message_id}")
return False
# Handle message IDs - they come as bytes and need to be decoded
raw_msg_ids = search_response.lines[0].split()
msg_ids = []
for raw_id in raw_msg_ids:
if isinstance(raw_id, bytes):
msg_ids.append(raw_id.decode('utf-8'))
else:
msg_ids.append(str(raw_id))
print(f"Found {len(msg_ids)} messages to mark as read")
for msg_id in msg_ids:
await self.client.store(msg_id, "+FLAGS", "\\Seen")
return True
except Exception as e:
print(f"Failed to mark message as read: {e}")
raise RuntimeError(f"Failed to mark message as read: {e}") from e
class EmailService:
"""Main email service combining SMTP and IMAP functionality."""
def __init__(self):
self.config = EmailConfig()
self.smtp_client = SMTPClient(self.config)
self.imap_client = IMAPClient(self.config)
async def send_email(self, *args, **kwargs) -> bool:
"""Send email via SMTP client."""
return await self.smtp_client.send_email(*args, **kwargs)
async def send_game_notification(self, *args, **kwargs) -> bool:
"""Send game notification via SMTP client."""
return await self.smtp_client.send_game_notification(*args, **kwargs)
async def fetch_new_messages(self, *args, **kwargs) -> List[EmailMessage]:
"""Fetch new messages via IMAP client."""
return await self.imap_client.fetch_new_messages(*args, **kwargs)
async def mark_as_read(self, *args, **kwargs) -> bool:
"""Mark message as read via IMAP client."""
return await self.imap_client.mark_as_read(*args, **kwargs)
async def process_incoming_emails(self) -> List[Dict[str, Any]]:
"""Process incoming emails and convert them to game actions."""
messages = await self.fetch_new_messages(mark_as_read=True)
processed_messages = []
for msg in messages:
# Parse game commands from email content
processed_msg = {
"from_email": msg.from_email,
"subject": msg.subject,
"content": msg.content,
"message_id": msg.message_id,
"received_at": msg.received_at,
"commands": self._extract_game_commands(msg.content)
}
processed_messages.append(processed_msg)
return processed_messages
def _extract_game_commands(self, content: str) -> List[str]:
"""Extract game commands from email content."""
# Simple command extraction - can be enhanced
commands = []
lines = content.split('\n')
for line in lines:
line = line.strip().lower()
# Look for common game commands
if line.startswith(('go ', 'move ', 'walk ', 'run ')):
commands.append(line)
elif line.startswith(('say ', 'tell ', 'whisper ')):
commands.append(line)
elif line.startswith(('look', 'examine ', 'inspect ')):
commands.append(line)
elif line.startswith(('take ', 'get ', 'pick up ')):
commands.append(line)
elif line.startswith(('drop ', 'put down ')):
commands.append(line)
elif line.startswith(('use ', 'cast ', 'drink ', 'eat ')):
commands.append(line)
elif line in ['help', 'inventory', 'stats', 'status', 'quit', 'save']:
commands.append(line)
return commands
async def cleanup(self):
"""Cleanup email connections."""
await self.imap_client.disconnect()

28
src/carbon_copy/game.py Normal file
View file

@ -0,0 +1,28 @@
from carbon_copy.player import Player, Room
from carbon_copy.database import Database
from dataclasses import dataclass
from contextlib import contextmanager
from typing import Any
__all__ = "Game",
class Game:
def __init__(self) -> None:
self.database = Database()
async def start(self) -> None:
...
async def stop(self) -> None:
...
async def __aenter__(self) -> Game:
await self.start()
return self
async def __aexit__(self, *_: Any) -> None:
await self.stop()
async def authenticate(self, name: str) -> Player:
pass

249
src/carbon_copy/main.py
View file

@ -1,249 +0,0 @@
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone
from pathlib import Path
from fastapi import Depends, FastAPI, Request
from fastapi.templating import Jinja2Templates
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from sqlalchemy.sql import func
from .database import create_db_and_tables, get_async_session
from .email_client import EmailService
from .models import * # Import all models
# Get the directory where this file is located
BASE_DIR = Path(__file__).resolve().parent
# Initialize email service
email_service = EmailService()
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan manager."""
# Startup
create_db_and_tables()
print("Database tables created")
print("Carbon Copy started - Email text adventure ready!")
yield
# Shutdown
await email_service.cleanup()
print("Carbon Copy shutting down")
app = FastAPI(
title="Carbon Copy",
description="An email-based multiplayer text adventure for the masses",
version="0.0.1",
lifespan=lifespan
)
templates = Jinja2Templates(directory=BASE_DIR / "templates")
@app.get("/")
async def root(request: Request):
"""Return the home page"""
return templates.TemplateResponse("index.html", {"request": request})
@app.get("/api/health")
async def health_check():
"""Health check endpoint."""
return {"status": "healthy", "service": "carbon-copy"}
@app.get("/api/stats")
async def get_stats(session: AsyncSession = Depends(get_async_session)):
"""Get basic game statistics."""
from sqlmodel import func, select
# Count users
user_result = await session.execute(select(func.count()).select_from(User))
user_count = user_result.scalar_one()
# Count rooms
room_result = await session.execute(select(func.count()).select_from(Room))
room_count = room_result.scalar_one()
# Count active users (online in last hour)
one_hour_ago = datetime.now(timezone.utc) - timedelta(hours=1)
try:
active_result = await session.execute(
select(func.count()).select_from(User).where(
User.last_activity.is_not(None) and User.last_activity > one_hour_ago # type: ignore
)
)
active_count = active_result.scalar_one()
except Exception:
# Fallback if the query fails
active_count = 0
return {
"total_users": user_count,
"total_rooms": room_count,
"active_users": active_count,
"game_status": "running"
}
@app.post("/api/process-emails")
async def process_emails(session: AsyncSession = Depends(get_async_session)):
"""Manually trigger email processing and save to database."""
try:
# Fetch and process emails
messages = await email_service.process_incoming_emails()
saved_messages = []
new_users_created = 0
for email_data in messages:
# Extract email address from the from_email field
from_email = email_data["from_email"]
# Parse email address (handle "Name <email@domain.com>" format)
if "<" in from_email and ">" in from_email:
email_address = from_email.split("<")[1].split(">")[0].strip()
else:
email_address = from_email.strip()
# Find or create user
user_result = await session.execute(
select(User).where(User.email == email_address)
)
user = user_result.scalar_one_or_none()
if not user:
# Create new user
username = email_address.split("@")[0] # Use part before @ as username
# Make sure username is unique
existing_username = await session.execute(
select(User).where(User.username == username)
)
if existing_username.scalar_one_or_none():
# Add timestamp to make it unique
import time
username = f"{username}_{int(time.time())}"
user = User(
username=username,
email=email_address,
display_name=username,
hashed_password="email_user_no_password", # Placeholder for email-only users
role=UserRole.PLAYER,
status=UserStatus.ACTIVE
)
session.add(user)
await session.commit() # Commit to get the user ID
await session.refresh(user)
new_users_created += 1
print(f"Created new user: {username} ({email_address})")
# Create message record
message = Message(
subject=email_data["subject"],
content=email_data["content"],
message_type=MessageType.CHAT,
status=MessageStatus.DELIVERED,
sender_id=user.id,
email_message_id=email_data["message_id"],
contains_commands=len(email_data["commands"]) > 0,
created_at=email_data["received_at"] or datetime.now(timezone.utc),
delivered_at=datetime.now(timezone.utc)
)
session.add(message)
# Update user's last activity
user.last_activity = datetime.now(timezone.utc)
user.last_email_check = datetime.now(timezone.utc)
saved_messages.append({
"id": "pending", # Will be set after commit
"from_email": email_address,
"subject": email_data["subject"],
"content": email_data["content"][:100] + "..." if len(email_data["content"]) > 100 else email_data["content"],
"commands": email_data["commands"],
"user_id": user.id,
"username": user.username
})
# Commit all changes
await session.commit()
return {
"status": "success",
"messages_processed": len(messages),
"messages_saved": len(saved_messages),
"new_users_created": new_users_created,
"saved_messages": saved_messages
}
except Exception as e:
await session.rollback()
print(f"Error processing emails: {e}")
return {
"status": "error",
"message": str(e)
}
@app.get("/api/messages")
async def get_messages(
limit: int = 20,
offset: int = 0,
session: AsyncSession = Depends(get_async_session)
):
"""Get recent messages from the database."""
try:
# Get recent email messages (all messages in this system are from emails)
messages_result = await session.execute(
select(Message)
.limit(limit)
.offset(offset)
)
messages_data = messages_result.scalars().all()
# Get all users for lookup
users_result = await session.execute(select(User))
users = {user.id: user for user in users_result.scalars().all()}
messages = []
for message in messages_data:
sender = users.get(message.sender_id) if message.sender_id else None
messages.append({
"id": message.id,
"subject": message.subject,
"content": message.content,
"email_message_id": message.email_message_id,
"contains_commands": message.contains_commands,
"created_at": message.created_at.isoformat() if message.created_at else None,
"sender": {
"id": sender.id if sender else None,
"username": sender.username if sender else None,
"email": sender.email if sender else None,
"display_name": sender.display_name if sender else None
} if sender else None
})
return {
"status": "success",
"messages": messages,
"limit": limit,
"offset": offset,
"count": len(messages)
}
except Exception as e:
return {
"status": "error",
"message": str(e)
}

503
src/carbon_copy/models.py
View file

@ -1,503 +0,0 @@
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
from datetime import datetime, timezone
from typing import Optional, List
from enum import Enum
from sqlmodel import SQLModel, Field, Relationship
class UserStatus(str, Enum):
ACTIVE = "active"
INACTIVE = "inactive"
BANNED = "banned"
SUSPENDED = "suspended"
class UserRole(str, Enum):
PLAYER = "player"
MODERATOR = "moderator"
ADMIN = "admin"
GAME_MASTER = "game_master"
class MessageType(str, Enum):
CHAT = "chat"
ACTION = "action"
SYSTEM = "system"
WHISPER = "whisper"
ROOM_ANNOUNCEMENT = "room_announcement"
GLOBAL_ANNOUNCEMENT = "global_announcement"
class MessageStatus(str, Enum):
SENT = "sent"
DELIVERED = "delivered"
READ = "read"
FAILED = "failed"
class ItemType(str, Enum):
WEAPON = "weapon"
ARMOR = "armor"
CONSUMABLE = "consumable"
TOOL = "tool"
TREASURE = "treasure"
KEY = "key"
BOOK = "book"
CONTAINER = "container"
QUEST_ITEM = "quest_item"
class ItemRarity(str, Enum):
COMMON = "common"
UNCOMMON = "uncommon"
RARE = "rare"
EPIC = "epic"
LEGENDARY = "legendary"
class RoomType(str, Enum):
INDOOR = "indoor"
OUTDOOR = "outdoor"
DUNGEON = "dungeon"
WILDERNESS = "wilderness"
CITY = "city"
SHOP = "shop"
TAVERN = "tavern"
TEMPLE = "temple"
class NPCType(str, Enum):
MERCHANT = "merchant"
GUARD = "guard"
QUEST_GIVER = "quest_giver"
ENEMY = "enemy"
FRIENDLY = "friendly"
NEUTRAL = "neutral"
BOSS = "boss"
class VehicleType(str, Enum):
HORSE = "horse"
CART = "cart"
SHIP = "ship"
CARRIAGE = "carriage"
FLYING_MOUNT = "flying_mount"
MAGICAL_TRANSPORT = "magical_transport"
class VehicleStatus(str, Enum):
AVAILABLE = "available"
IN_USE = "in_use"
BROKEN = "broken"
MAINTENANCE = "maintenance"
# Association tables for many-to-many relationships
class UserItemLink(SQLModel, table=True):
user_id: Optional[int] = Field(default=None, foreign_key="user.id", primary_key=True)
item_id: Optional[int] = Field(default=None, foreign_key="item.id", primary_key=True)
quantity: int = Field(default=1)
equipped: bool = Field(default=False)
obtained_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
class RoomItemLink(SQLModel, table=True):
room_id: Optional[int] = Field(default=None, foreign_key="room.id", primary_key=True)
item_id: Optional[int] = Field(default=None, foreign_key="item.id", primary_key=True)
quantity: int = Field(default=1)
position_x: Optional[float] = Field(default=None)
position_y: Optional[float] = Field(default=None)
hidden: bool = Field(default=False)
class UserRoomHistory(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
user_id: int = Field(foreign_key="user.id")
room_id: int = Field(foreign_key="room.id")
entered_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
left_at: Optional[datetime] = Field(default=None)
duration_minutes: Optional[int] = Field(default=None)
# Relationships
user: Optional["User"] = Relationship(back_populates="room_history")
# Main models
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
email: str = Field(unique=True, index=True)
username: str = Field(unique=True, index=True)
display_name: str
hashed_password: str
# Game stats
level: int = Field(default=1)
experience: int = Field(default=0)
health: int = Field(default=100)
max_health: int = Field(default=100)
energy: int = Field(default=100)
max_energy: int = Field(default=100)
strength: int = Field(default=10)
dexterity: int = Field(default=10)
intelligence: int = Field(default=10)
charisma: int = Field(default=10)
# Currency and resources
gold: int = Field(default=0)
silver: int = Field(default=0)
copper: int = Field(default=100)
# Status and role
status: UserStatus = Field(default=UserStatus.ACTIVE)
role: UserRole = Field(default=UserRole.PLAYER)
# Location and movement
current_room_id: Optional[int] = Field(default=1, foreign_key="room.id")
last_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
position_x: Optional[float] = Field(default=0.0)
position_y: Optional[float] = Field(default=0.0)
# Timestamps
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
last_login: Optional[datetime] = Field(default=None)
last_activity: Optional[datetime] = Field(default=None)
last_email_check: Optional[datetime] = Field(default=None)
# Game state
is_online: bool = Field(default=False)
is_in_combat: bool = Field(default=False)
is_resting: bool = Field(default=False)
respawn_time: Optional[datetime] = Field(default=None)
# Preferences
email_notifications: bool = Field(default=True)
auto_save_interval: int = Field(default=300) # seconds
preferred_language: str = Field(default="en")
timezone: str = Field(default="UTC")
# Relationships
current_room: Optional["Room"] = Relationship(
back_populates="current_users",
sa_relationship_kwargs={"foreign_keys": "[User.current_room_id]"}
)
sent_messages: List["Message"] = Relationship(
back_populates="sender",
sa_relationship_kwargs={"foreign_keys": "[Message.sender_id]"}
)
received_messages: List["Message"] = Relationship(
back_populates="recipient",
sa_relationship_kwargs={"foreign_keys": "[Message.recipient_id]"}
)
items: List["Item"] = Relationship(back_populates="users", link_model=UserItemLink)
room_history: List["UserRoomHistory"] = Relationship(back_populates="user")
owned_vehicles: List["Vehicle"] = Relationship(back_populates="owner")
class Room(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
short_description: str
long_description: str
# Room properties
room_type: RoomType = Field(default=RoomType.INDOOR)
is_safe: bool = Field(default=True)
is_pvp_enabled: bool = Field(default=False)
is_accessible: bool = Field(default=True)
is_hidden: bool = Field(default=False)
requires_light: bool = Field(default=False)
# Environmental factors
temperature: Optional[int] = Field(default=20) # Celsius
humidity: Optional[int] = Field(default=50) # Percentage
light_level: int = Field(default=100) # 0-100
noise_level: int = Field(default=0) # 0-100
# Size and capacity
max_occupancy: int = Field(default=10)
area_square_meters: Optional[float] = Field(default=None)
ceiling_height: Optional[float] = Field(default=None)
# Coordinates for mapping
zone_id: Optional[int] = Field(default=None)
world_x: Optional[float] = Field(default=None)
world_y: Optional[float] = Field(default=None)
world_z: Optional[float] = Field(default=None)
# Connected rooms (simplified - could be expanded to a separate table)
north_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
south_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
east_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
west_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
up_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
down_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
# Timestamps
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
last_cleaned: Optional[datetime] = Field(default=None)
# Relationships
current_users: List["User"] = Relationship(
back_populates="current_room",
sa_relationship_kwargs={"foreign_keys": "[User.current_room_id]"}
)
messages: List["Message"] = Relationship(back_populates="room")
items: List["Item"] = Relationship(back_populates="rooms", link_model=RoomItemLink)
npcs: List["NPC"] = Relationship(
back_populates="current_room",
sa_relationship_kwargs={"foreign_keys": "[NPC.current_room_id]"}
)
vehicles: List["Vehicle"] = Relationship(
back_populates="current_room",
sa_relationship_kwargs={"foreign_keys": "[Vehicle.current_room_id]"}
)
class Message(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
subject: Optional[str] = Field(default=None)
content: str
# Message metadata
message_type: MessageType = Field(default=MessageType.CHAT)
status: MessageStatus = Field(default=MessageStatus.SENT)
priority: int = Field(default=1) # 1-5, 5 being highest
# Sender and recipient
sender_id: Optional[int] = Field(default=None, foreign_key="user.id")
recipient_id: Optional[int] = Field(default=None, foreign_key="user.id")
room_id: Optional[int] = Field(default=None, foreign_key="room.id")
# Email integration
email_message_id: Optional[str] = Field(default=None, unique=True)
email_thread_id: Optional[str] = Field(default=None)
reply_to_message_id: Optional[int] = Field(default=None, foreign_key="message.id")
# Timestamps
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
sent_at: Optional[datetime] = Field(default=None)
delivered_at: Optional[datetime] = Field(default=None)
read_at: Optional[datetime] = Field(default=None)
expires_at: Optional[datetime] = Field(default=None)
# Content flags
is_encrypted: bool = Field(default=False)
contains_commands: bool = Field(default=False)
is_system_generated: bool = Field(default=False)
is_broadcast: bool = Field(default=False)
# Relationships
sender: Optional["User"] = Relationship(
back_populates="sent_messages",
sa_relationship_kwargs={"foreign_keys": "[Message.sender_id]"}
)
recipient: Optional["User"] = Relationship(
back_populates="received_messages",
sa_relationship_kwargs={"foreign_keys": "[Message.recipient_id]"}
)
room: Optional["Room"] = Relationship(back_populates="messages")
replies: List["Message"] = Relationship(
back_populates="parent_message",
sa_relationship_kwargs={"foreign_keys": "[Message.reply_to_message_id]"}
)
parent_message: Optional["Message"] = Relationship(
back_populates="replies",
sa_relationship_kwargs={"remote_side": "[Message.id]", "foreign_keys": "[Message.reply_to_message_id]"}
)
class Item(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
description: str
short_description: str
# Item properties
item_type: ItemType = Field(default=ItemType.TREASURE)
rarity: ItemRarity = Field(default=ItemRarity.COMMON)
# Value and weight
base_value: int = Field(default=0) # in copper
weight: float = Field(default=0.0) # in kg
volume: float = Field(default=0.0) # in liters
# Durability and condition
max_durability: int = Field(default=100)
current_durability: int = Field(default=100)
repair_cost: int = Field(default=0)
# Combat stats (for weapons/armor)
damage: int = Field(default=0)
armor_value: int = Field(default=0)
accuracy_bonus: int = Field(default=0)
critical_chance: float = Field(default=0.0)
# Stat bonuses
strength_bonus: int = Field(default=0)
dexterity_bonus: int = Field(default=0)
intelligence_bonus: int = Field(default=0)
charisma_bonus: int = Field(default=0)
health_bonus: int = Field(default=0)
energy_bonus: int = Field(default=0)
# Usage properties
is_consumable: bool = Field(default=False)
is_stackable: bool = Field(default=False)
max_stack_size: int = Field(default=1)
uses_remaining: Optional[int] = Field(default=None)
cooldown_seconds: int = Field(default=0)
# Requirements
required_level: int = Field(default=1)
required_strength: int = Field(default=0)
required_dexterity: int = Field(default=0)
required_intelligence: int = Field(default=0)
# Flags
is_unique: bool = Field(default=False)
is_cursed: bool = Field(default=False)
is_blessed: bool = Field(default=False)
is_magical: bool = Field(default=False)
is_quest_item: bool = Field(default=False)
can_be_dropped: bool = Field(default=True)
can_be_traded: bool = Field(default=True)
can_be_sold: bool = Field(default=True)
# Timestamps
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
# Relationships
users: List["User"] = Relationship(back_populates="items", link_model=UserItemLink)
rooms: List["Room"] = Relationship(back_populates="items", link_model=RoomItemLink)
class NPC(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
description: str
short_description: str
# NPC properties
npc_type: NPCType = Field(default=NPCType.FRIENDLY)
level: int = Field(default=1)
# Stats
health: int = Field(default=100)
max_health: int = Field(default=100)
energy: int = Field(default=100)
max_energy: int = Field(default=100)
strength: int = Field(default=10)
dexterity: int = Field(default=10)
intelligence: int = Field(default=10)
charisma: int = Field(default=10)
# Combat properties
damage: int = Field(default=10)
armor: int = Field(default=0)
accuracy: int = Field(default=75)
evasion: int = Field(default=5)
# Behavior
is_aggressive: bool = Field(default=False)
is_mobile: bool = Field(default=True)
respawn_time_minutes: int = Field(default=15)
patrol_range: int = Field(default=3) # rooms
# Location
current_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
spawn_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
position_x: Optional[float] = Field(default=0.0)
position_y: Optional[float] = Field(default=0.0)
# Loot and rewards
gold_drop_min: int = Field(default=0)
gold_drop_max: int = Field(default=0)
experience_reward: int = Field(default=10)
# Dialogue and quests
greeting_message: Optional[str] = Field(default=None)
farewell_message: Optional[str] = Field(default=None)
shop_inventory: Optional[str] = Field(default=None) # JSON string of item IDs
# State
is_alive: bool = Field(default=True)
last_death: Optional[datetime] = Field(default=None)
last_respawn: Optional[datetime] = Field(default=None)
last_movement: Optional[datetime] = Field(default=None)
# Timestamps
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
# Relationships
current_room: Optional["Room"] = Relationship(
back_populates="npcs",
sa_relationship_kwargs={"foreign_keys": "[NPC.current_room_id]"}
)
class Vehicle(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
description: str
# Vehicle properties
vehicle_type: VehicleType = Field(default=VehicleType.HORSE)
status: VehicleStatus = Field(default=VehicleStatus.AVAILABLE)
# Capacity and size
max_passengers: int = Field(default=1)
max_cargo_weight: float = Field(default=100.0) # kg
max_cargo_volume: float = Field(default=50.0) # liters
# Performance
speed: float = Field(default=1.0) # multiplier for travel time
fuel_capacity: Optional[float] = Field(default=None)
fuel_current: Optional[float] = Field(default=None)
fuel_consumption: Optional[float] = Field(default=None) # per distance unit
# Durability
max_durability: int = Field(default=100)
current_durability: int = Field(default=100)
maintenance_cost: int = Field(default=0)
last_maintenance: Optional[datetime] = Field(default=None)
# Location and ownership
owner_id: Optional[int] = Field(default=None, foreign_key="user.id")
current_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
parked_room_id: Optional[int] = Field(default=None, foreign_key="room.id")
# Requirements
required_skill_level: int = Field(default=0)
license_required: bool = Field(default=False)
# Value
purchase_cost: int = Field(default=1000)
current_value: int = Field(default=1000)
insurance_cost: int = Field(default=0)
# Features
has_storage: bool = Field(default=False)
is_magical: bool = Field(default=False)
can_fly: bool = Field(default=False)
can_sail: bool = Field(default=False)
is_amphibious: bool = Field(default=False)
# Timestamps
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
last_used: Optional[datetime] = Field(default=None)
# Relationships
owner: Optional["User"] = Relationship(back_populates="owned_vehicles")
current_room: Optional["Room"] = Relationship(
back_populates="vehicles",
sa_relationship_kwargs={"foreign_keys": "[Vehicle.current_room_id]"}
)

73
src/carbon_copy/player.py Normal file
View file

@ -0,0 +1,73 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import NewType, Self
from enum import Enum, auto
import httpx
from carbon_copy.database import Database, DatabaseItem, UUID
__all__ = "Player",
class AIInteraction:
def __init__(self, endpoint: str = "https://ai.hackclub.com/chat/completions") -> None:
self.endpoint = endpoint
async def interact(self, system_prompt: str, user_prompt: str):
async with httpx.AsyncClient() as client:
response = await client.post(self.endpoint, headers={"Content-Type": "application/json"}, json={
"messages": [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}]
})
if response.status_code != 200:
print(response.text)
raise ValueError(f"Interaction with AI returned bad HTTP status code {response.status_code}!")
else:
return response.json()
@dataclass(slots=True)
class Room(DatabaseItem):
description: str
parent_room: Room
child_rooms: list[Room]
@classmethod
async def generate_room(cls) -> Room:
...
@classmethod
def lobby(cls) -> Room:
...
class ItemType(Enum):
WEAPON = auto()
ARMOR = auto()
CONSUMABLE = auto()
RESOURCE = auto()
@dataclass(slots=True)
class Item(DatabaseItem):
description: str
item_type: ItemType
@classmethod
async def generate_item(cls) -> Item:
...
@dataclass(slots=True)
class Response:
content: str
options: list[str]
@dataclass(slots=True)
class Player(DatabaseItem):
room: Room
items: list[Item]
def play(self, action: str) -> Response:
...

23
src/carbon_copy/resp.json Normal file
View file

@ -0,0 +1,23 @@
{
"type": "weapon",
"name": "Celestial Judgment Halberd",
"description": "A massive polearm forged from star-metal by an artisan long thought dead. Its blade shimmers with an ethereal light and pulses with cosmic energy that repels evil and enhances the wielder's power.",
"weaponCategory": "Halberd",
"damageType": "Slashing",
"rarity": "Legendary",
"weight": 8,
"baseDamage": 12,
"modifier": "+3 to hit",
"durability": {
"current": 50,
"max": 50
},
"range": 1.5,
"specialAbilities": [
"Celestial Judgment: On a critical hit, releases a burst of radiant energy dealing 20 points of light damage to all adjacent enemies and heals the wielder for 20 hit points",
"Divine Resonance: Deals +10 damage to creatures corrupted by darkness or undead",
"Starforge Bond: Gains temporary hit points equal to twice the wielder's level at the start of each short rest"
],
"value": 2000,
"lore": "Crafted from meteoric iron that fell during the Great Skyfire, this weapon was wielded by a celestial champion who ended a thousand-year-old demonic infestation. It hums softly when near portals to the Shadowplane."
}

327
src/carbon_copy/seed_data.py
View file

@ -1,327 +0,0 @@
# SPDX-FileCopyrightText: 2025-present Micha Albert <info@micha.zone>
#
# SPDX-License-Identifier: MIT
from datetime import datetime, timezone
from sqlmodel import Session, select
from .database import engine
from .models import *
def create_initial_rooms():
"""Create initial game rooms."""
rooms_data = [
{
"id": 1,
"name": "Town Square",
"short_description": "The bustling heart of the town",
"long_description": "You stand in the center of a busy town square. Cobblestone paths lead in all directions, and a beautiful fountain sits in the middle. Merchants hawk their wares while children play around the fountain's edge.",
"room_type": RoomType.OUTDOOR,
"is_safe": True,
"max_occupancy": 20,
"north_room_id": 2,
"east_room_id": 3,
"west_room_id": 4,
"south_room_id": 5
},
{
"id": 2,
"name": "North Gate",
"short_description": "The northern entrance to town",
"long_description": "A massive stone archway marks the northern entrance to the town. Guards in shining armor stand watch, checking travelers as they come and go. Beyond the gate, you can see a winding road leading into dark woods.",
"room_type": RoomType.OUTDOOR,
"is_safe": True,
"south_room_id": 1,
"north_room_id": 6
},
{
"id": 3,
"name": "The Prancing Pony Tavern",
"short_description": "A cozy tavern filled with warm light",
"long_description": "The Prancing Pony is a welcoming tavern with low wooden beams, flickering candles, and the smell of roasted meat in the air. Patrons sit around heavy oak tables, sharing stories and ale. A large fireplace crackles merrily in the corner.",
"room_type": RoomType.TAVERN,
"is_safe": True,
"west_room_id": 1,
"max_occupancy": 15
},
{
"id": 4,
"name": "General Store",
"short_description": "A well-stocked merchant's shop",
"long_description": "Shelves line the walls from floor to ceiling, packed with all manner of goods. Weapons, armor, potions, and supplies fill every available space. The shopkeeper, a portly man with a friendly smile, stands behind a wooden counter.",
"room_type": RoomType.SHOP,
"is_safe": True,
"east_room_id": 1,
"max_occupancy": 8
},
{
"id": 5,
"name": "South Road",
"short_description": "A dusty road leading south",
"long_description": "The road stretches south from the town, disappearing over rolling hills. Cart tracks mark the packed earth, and wildflowers grow along the roadside. A wooden signpost points toward distant cities.",
"room_type": RoomType.OUTDOOR,
"is_safe": True,
"north_room_id": 1,
"south_room_id": 7
},
{
"id": 6,
"name": "Dark Woods",
"short_description": "A mysterious forest path",
"long_description": "Ancient trees tower overhead, their branches forming a canopy that blocks most of the sunlight. The air is cool and damp, filled with the sounds of rustling leaves and distant bird calls. A narrow path winds deeper into the woods.",
"room_type": RoomType.WILDERNESS,
"is_safe": False,
"south_room_id": 2,
"light_level": 30,
"requires_light": True
},
{
"id": 7,
"name": "Hillside Farm",
"short_description": "A peaceful farming settlement",
"long_description": "Rolling green fields stretch as far as the eye can see, dotted with grazing sheep and golden wheat. A small farmhouse sits atop a nearby hill, smoke rising from its chimney. The air smells of fresh earth and growing things.",
"room_type": RoomType.OUTDOOR,
"is_safe": True,
"north_room_id": 5,
"max_occupancy": 6
}
]
with Session(engine) as session:
for room_data in rooms_data:
# Check if room already exists
existing_room = session.exec(select(Room).where(Room.id == room_data["id"])).first()
if not existing_room:
room = Room(**room_data)
session.add(room)
session.commit()
def create_initial_items():
"""Create initial game items."""
items_data = [
{
"name": "Rusty Sword",
"description": "An old iron sword with spots of rust along the blade. Despite its age, it still holds a sharp edge.",
"short_description": "A rusty but functional sword",
"item_type": ItemType.WEAPON,
"rarity": ItemRarity.COMMON,
"base_value": 50,
"weight": 2.5,
"damage": 8,
"required_level": 1,
"can_be_traded": True
},
{
"name": "Leather Armor",
"description": "Well-crafted leather armor that provides basic protection while allowing good mobility.",
"short_description": "Simple but effective leather armor",
"item_type": ItemType.ARMOR,
"rarity": ItemRarity.COMMON,
"base_value": 75,
"weight": 5.0,
"armor_value": 5,
"required_level": 1,
"can_be_traded": True
},
{
"name": "Health Potion",
"description": "A small glass vial filled with a red liquid that glows with magical energy. Drinking it will restore some health.",
"short_description": "A magical healing potion",
"item_type": ItemType.CONSUMABLE,
"rarity": ItemRarity.COMMON,
"base_value": 25,
"weight": 0.2,
"health_bonus": 25,
"is_consumable": True,
"is_stackable": True,
"max_stack_size": 10,
"uses_remaining": 1
},
{
"name": "Traveler's Backpack",
"description": "A sturdy leather backpack with multiple compartments and reinforced straps. Perfect for long journeys.",
"short_description": "A reliable traveler's backpack",
"item_type": ItemType.CONTAINER,
"rarity": ItemRarity.COMMON,
"base_value": 40,
"weight": 1.0,
"can_be_traded": True
},
{
"name": "Magic Crystal",
"description": "A beautiful crystal that pulses with inner light. It feels warm to the touch and seems to contain powerful magic.",
"short_description": "A glowing magical crystal",
"item_type": ItemType.TREASURE,
"rarity": ItemRarity.RARE,
"base_value": 500,
"weight": 0.5,
"is_magical": True,
"intelligence_bonus": 2,
"energy_bonus": 10
}
]
with Session(engine) as session:
for item_data in items_data:
# Check if item already exists
existing_item = session.exec(select(Item).where(Item.name == item_data["name"])).first()
if not existing_item:
item = Item(**item_data)
session.add(item)
session.commit()
def create_initial_npcs():
"""Create initial NPCs."""
npcs_data = [
{
"name": "Marcus the Shopkeeper",
"description": "A portly, middle-aged man with a friendly disposition and keen eye for business.",
"short_description": "The friendly general store owner",
"npc_type": NPCType.MERCHANT,
"level": 5,
"current_room_id": 4,
"spawn_room_id": 4,
"is_aggressive": False,
"is_mobile": False,
"greeting_message": "Welcome to my shop! I have the finest goods in town.",
"farewell_message": "Thank you for your business! Come back anytime.",
"charisma": 15
},
{
"name": "Captain Roderick",
"description": "A stern-looking guard captain with graying hair and battle scars. His armor is well-maintained and his sword always ready.",
"short_description": "The town guard captain",
"npc_type": NPCType.GUARD,
"level": 8,
"current_room_id": 2,
"spawn_room_id": 2,
"strength": 16,
"dexterity": 12,
"health": 150,
"max_health": 150,
"damage": 15,
"armor": 8,
"is_aggressive": False,
"greeting_message": "Halt! State your business in our fair town.",
"farewell_message": "Move along, citizen. Keep the peace."
},
{
"name": "Old Tom the Storyteller",
"description": "An elderly man with twinkling eyes and a white beard. He sits by the tavern's fireplace, always ready with a tale.",
"short_description": "An old storyteller with many tales",
"npc_type": NPCType.QUEST_GIVER,
"level": 3,
"current_room_id": 3,
"spawn_room_id": 3,
"is_aggressive": False,
"is_mobile": False,
"intelligence": 18,
"charisma": 16,
"greeting_message": "Come, sit by the fire! I have stories that will amaze you.",
"farewell_message": "Safe travels, young one. May your adventures be legendary!"
},
{
"name": "Forest Wolf",
"description": "A large, gray wolf with yellow eyes that gleam with wild intelligence. Its hackles are raised and it bares its fangs.",
"short_description": "A dangerous forest predator",
"npc_type": NPCType.ENEMY,
"level": 4,
"current_room_id": 6,
"spawn_room_id": 6,
"strength": 14,
"dexterity": 16,
"health": 80,
"max_health": 80,
"damage": 12,
"armor": 2,
"accuracy": 80,
"evasion": 15,
"is_aggressive": True,
"respawn_time_minutes": 30,
"experience_reward": 50,
"gold_drop_min": 5,
"gold_drop_max": 15
}
]
with Session(engine) as session:
for npc_data in npcs_data:
# Check if NPC already exists
existing_npc = session.exec(select(NPC).where(NPC.name == npc_data["name"])).first()
if not existing_npc:
npc = NPC(**npc_data)
session.add(npc)
session.commit()
def create_initial_vehicles():
"""Create initial vehicles."""
vehicles_data = [
{
"name": "Brown Mare",
"description": "A sturdy brown horse with a gentle temperament. She's well-trained for riding and responds well to commands.",
"vehicle_type": VehicleType.HORSE,
"status": VehicleStatus.AVAILABLE,
"max_passengers": 1,
"max_cargo_weight": 50.0,
"speed": 2.0,
"current_room_id": 1,
"parked_room_id": 1,
"purchase_cost": 300,
"current_value": 300,
"max_durability": 100,
"current_durability": 100
},
{
"name": "Merchant's Cart",
"description": "A wooden cart with iron-reinforced wheels. Perfect for transporting goods between towns.",
"vehicle_type": VehicleType.CART,
"status": VehicleStatus.AVAILABLE,
"max_passengers": 2,
"max_cargo_weight": 200.0,
"max_cargo_volume": 150.0,
"speed": 1.5,
"current_room_id": 4,
"parked_room_id": 4,
"purchase_cost": 150,
"current_value": 150,
"has_storage": True,
"required_skill_level": 1
}
]
with Session(engine) as session:
for vehicle_data in vehicles_data:
# Check if vehicle already exists
existing_vehicle = session.exec(select(Vehicle).where(Vehicle.name == vehicle_data["name"])).first()
if not existing_vehicle:
vehicle = Vehicle(**vehicle_data)
session.add(vehicle)
session.commit()
def initialize_game_world():
"""Initialize the complete game world with all seed data."""
print("Creating initial rooms...")
create_initial_rooms()
print("Creating initial items...")
create_initial_items()
print("Creating initial NPCs...")
create_initial_npcs()
print("Creating initial vehicles...")
create_initial_vehicles()
print("Game world initialization complete!")
if __name__ == "__main__":
initialize_game_world()

32
src/carbon_copy/templates/index.html
View file

@ -1,32 +0,0 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta name="description" content="An email-based multiplayer text adventure for the masses">
<title>Carbon Copy</title>
</head>
<body>
<h1>Welcome to Carbon Copy</h1>
<p>An email-based multiplayer text adventure for the masses</p>
<h2>How to Play</h2>
<ul>
<li>Send an email to the game master to create your character</li>
<li>Use simple commands like "look", "go north", "take sword"</li>
<li>Receive game updates via email</li>
<li>Adventure with other players in a persistent world</li>
</ul>
<h2>Game Status</h2>
<p>The Carbon Copy text adventure is running and ready for players!</p>
<h3>Quick Commands</h3>
<ul>
<li><strong>Movement:</strong> go north, go south, go east, go west</li>
<li><strong>Interaction:</strong> look, examine [item], take [item], drop [item]</li>
<li><strong>Communication:</strong> say [message], tell [player] [message]</li>
<li><strong>Character:</strong> inventory, stats, help</li>
</ul>
</body>
</html>