aodhan/swiper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Basic UI

Browse Source
This commit is contained in:
Aodhan
2025-06-20 22:27:24 +01:00
parent c09461f58f
commit 324a21800a
10 changed files with 1022 additions and 1866 deletions
Download Patch File Download Diff File Expand all files Collapse all files

282
app.py
View File

@@ -21,53 +21,42 @@ def init_db():
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Check if orientation column exists
cursor.execute("PRAGMA table_info(image_selections)")
columns = [column[1] for column in cursor.fetchall()]
# Create table if it doesn't exist
if 'image_selections' not in [table[0] for table in cursor.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()]:
cursor.execute('''
CREATE TABLE IF NOT EXISTS image_selections (
id INTEGER PRIMARY KEY AUTOINCREMENT,
image_path TEXT NOT NULL,
resolution TEXT NOT NULL,
action TEXT NOT NULL,
timestamp INTEGER NOT NULL,
orientation TEXT
)
''')
print("Created new image_selections table with orientation column")
elif 'orientation' not in columns:
# Add orientation column if it doesn't exist
cursor.execute('ALTER TABLE image_selections ADD COLUMN orientation TEXT')
print("Added orientation column to existing table")
# Create image_selections table
cursor.execute('''
CREATE TABLE IF NOT EXISTS image_selections (
id INTEGER PRIMARY KEY AUTOINCREMENT,
image_path TEXT NOT NULL UNIQUE,
action TEXT NOT NULL,
timestamp INTEGER NOT NULL
)
''')
# Create image_metadata table
cursor.execute('''
CREATE TABLE IF NOT EXISTS image_metadata (
id INTEGER PRIMARY KEY AUTOINCREMENT,
path TEXT NOT NULL UNIQUE,
resolution TEXT NOT NULL,
name TEXT NOT NULL,
orientation TEXT NOT NULL,
discovered_at INTEGER NOT NULL
)
''')
conn.commit()
conn.close()
print(f"Database initialized at {DB_PATH}")
# Add a selection to the database
def add_selection(image_path, resolution, action):
# Determine if image is portrait or landscape
orientation = "unknown"
try:
from PIL import Image
full_path = os.path.join(IMAGE_DIR, image_path.replace('/images/', ''))
with Image.open(full_path) as img:
width, height = img.size
orientation = "portrait" if height > width else "landscape" if width > height else "square"
except Exception as e:
print(f"DEBUG ERROR determining image orientation: {str(e)}")
def add_selection(image_path, action):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Insert the selection
# Use REPLACE INTO to handle potential duplicates gracefully
cursor.execute('''
INSERT INTO image_selections (image_path, resolution, action, timestamp, orientation)
VALUES (?, ?, ?, ?, ?)
''', (image_path, resolution, action, int(time.time()), orientation))
REPLACE INTO image_selections (image_path, action, timestamp)
VALUES (?, ?, ?)
''', (image_path, action, int(time.time())))
conn.commit()
conn.close()
@@ -98,7 +87,6 @@ def get_selections():
if 'orientation' not in item or not item['orientation']:
try:
# Try to determine orientation if not in database
from PIL import Image
image_path = item['image_path']
if image_path.startswith('/images/'):
image_path = image_path[8:]
@@ -123,23 +111,65 @@ def get_selections():
return []
# Get a list of all image paths that have already been actioned
def get_actioned_images():
try:
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute('''
SELECT DISTINCT image_path FROM image_selections
''')
rows = cursor.fetchall()
actioned_images = [row[0] for row in rows]
def sync_image_database():
"""Scans the image directory and adds any new images to the metadata table."""
print("Syncing image database...")
from PIL import Image
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Get all image paths already in the database
cursor.execute("SELECT path FROM image_metadata")
db_images = {row[0] for row in cursor.fetchall()}
print(f"Found {len(db_images)} images in the database.")
# Find all images on the filesystem
disk_images = set()
resolutions = [d for d in os.listdir(IMAGE_DIR) if os.path.isdir(os.path.join(IMAGE_DIR, d))]
for res in resolutions:
res_dir = os.path.join(IMAGE_DIR, res)
for img_name in os.listdir(res_dir):
if img_name.lower().endswith(('.png', '.jpg', '.jpeg')):
disk_images.add(f"{res}/{img_name}")
print(f"Found {len(disk_images)} images on disk.")
# Determine which images are new
new_images = disk_images - db_images
print(f"Found {len(new_images)} new images to add to the database.")
if not new_images:
print("Database is already up-to-date.")
conn.close()
return actioned_images
except Exception as e:
print(f"DEBUG ERROR in get_actioned_images(): {str(e)}")
return []
return
# Process and add new images to the database
images_to_add = []
total_new_images = len(new_images)
processed_count = 0
for image_path in new_images:
res, img_name = image_path.split('/', 1)
full_path = os.path.join(IMAGE_DIR, image_path)
try:
with Image.open(full_path) as img:
width, height = img.size
orientation = 'landscape' if width >= height else 'portrait'
images_to_add.append((image_path, res, img_name, orientation, int(time.time())))
processed_count += 1
if processed_count % 100 == 0 or processed_count == total_new_images:
percentage = (processed_count / total_new_images) * 100
print(f"Processed {processed_count} of {total_new_images} images ({percentage:.2f}%)...", flush=True)
except Exception as e:
print(f"Could not process image {full_path}: {e}")
if images_to_add:
cursor.executemany('''
INSERT INTO image_metadata (path, resolution, name, orientation, discovered_at)
VALUES (?, ?, ?, ?, ?)
''', images_to_add)
conn.commit()
print(f"Successfully added {len(images_to_add)} new images to the database.")
conn.close()
# Update a selection in the database
def update_selection(selection_id, action):
@@ -197,6 +227,8 @@ def reset_database():
print(f"DEBUG: Reset database - deleted {rows_affected} selections")
return rows_affected
class ImageSwipeHandler(BaseHTTPRequestHandler):
# Set response headers for CORS
def _set_cors_headers(self):
@@ -233,6 +265,34 @@ class ImageSwipeHandler(BaseHTTPRequestHandler):
self.serve_file(path)
except:
self.send_error(404, "File not found")
def do_POST(self):
parsed_url = urllib.parse.urlparse(self.path)
path = parsed_url.path
if path == '/selection':
try:
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
data = json.loads(post_data)
add_selection(data['image_path'], data['action'])
self.send_response(200)
self.send_header('Content-type', 'application/json')
self._set_cors_headers()
self.end_headers()
self.wfile.write(json.dumps({'status': 'success'}).encode())
except Exception as e:
print(f"ERROR in do_POST /selection: {e}")
self.send_error(500, f"Server error processing selection: {e}")
else:
self.send_error(404, "Endpoint not found")
def do_OPTIONS(self):
self.send_response(204)
self._set_cors_headers()
self.end_headers()
def serve_file(self, file_path, content_type=None):
try:
@@ -283,75 +343,61 @@ class ImageSwipeHandler(BaseHTTPRequestHandler):
self.send_error(404, f"Image not found: {image_path}")
def serve_random_image(self):
print("DEBUG: serve_random_image() called")
try:
# Get list of already actioned images
actioned_images = get_actioned_images()
print(f"DEBUG: Found {len(actioned_images)} already actioned images")
# Get all resolution directories
resolutions = [d for d in os.listdir(IMAGE_DIR) if os.path.isdir(os.path.join(IMAGE_DIR, d))]
# Try to find an unactioned image
max_attempts = 20 # Limit the number of attempts to find an unactioned image
for attempt in range(max_attempts):
# Choose a random resolution
resolution = random.choice(resolutions)
resolution_dir = os.path.join(IMAGE_DIR, resolution)
# Get all images in the selected resolution directory
images = [f for f in os.listdir(resolution_dir) if f.endswith(('.png', '.jpg', '.jpeg'))]
if not images:
continue # Try another resolution if this one has no images
# Filter out already actioned images
unactioned_images = [img for img in images if f"{resolution}/{img}" not in actioned_images]
# If we have unactioned images, choose one randomly
if unactioned_images:
image_name = random.choice(unactioned_images)
print(f"DEBUG: Found unactioned image: {resolution}/{image_name}")
break
elif attempt == max_attempts - 1:
# If we've tried max_attempts times and still haven't found an unactioned image,
# just choose any image
image_name = random.choice(images)
print(f"DEBUG: No unactioned images found after {max_attempts} attempts, using: {resolution}/{image_name}")
else:
# This will only execute if the for loop completes without a break
# Choose any random image as fallback
resolution = random.choice(resolutions)
resolution_dir = os.path.join(IMAGE_DIR, resolution)
images = [f for f in os.listdir(resolution_dir) if f.endswith(('.png', '.jpg', '.jpeg'))]
if not images:
self.send_error(404, "No images found in any resolution directory")
return
image_name = random.choice(images)
print(f"DEBUG: Using fallback random image: {resolution}/{image_name}")
image_path = f"{resolution}/{image_name}"
parsed_url = urllib.parse.urlparse(self.path)
query_params = urllib.parse.parse_qs(parsed_url.query)
orientation_filter = query_params.get('orientation', ['all'])[0]
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Base query to get unactioned images
query = """
SELECT meta.path, meta.resolution, meta.name, meta.orientation
FROM image_metadata meta
LEFT JOIN image_selections sel ON meta.path = sel.image_path
WHERE sel.image_path IS NULL
"""
# Add orientation filter if specified
params = ()
if orientation_filter != 'all':
query += " AND meta.orientation = ?"
params = (orientation_filter,)
cursor.execute(query, params)
possible_images = cursor.fetchall()
conn.close()
if not possible_images:
print("DEBUG: No matching unactioned images found.")
response = {'message': 'No more images available for this filter.'}
self.send_response(200)
self.send_header('Content-type', 'application/json')
self._set_cors_headers()
self.end_headers()
self.wfile.write(json.dumps(response).encode())
return
# Choose one random image from the filtered list
chosen_image_row = random.choice(possible_images)
image_path = chosen_image_row[0]
resolution = chosen_image_row[1]
image_name = chosen_image_row[2]
orientation = chosen_image_row[3]
full_image_path = os.path.join(IMAGE_DIR, image_path)
# Get the file creation time
print(f"DEBUG: Serving image: {image_path}")
# Get file metadata
try:
file_stat = os.stat(full_image_path)
creation_time = file_stat.st_mtime # Use modification time as creation time
creation_time = file_stat.st_mtime
creation_date = datetime.datetime.fromtimestamp(creation_time).strftime('%Y-%m-%d %H:%M:%S')
except Exception as e:
print(f"DEBUG ERROR getting file creation time: {str(e)}")
creation_date = "Unknown"
# Determine if image is portrait or landscape
try:
from PIL import Image
with Image.open(full_image_path) as img:
width, height = img.size
orientation = "portrait" if height > width else "landscape" if width > height else "square"
except Exception as e:
print(f"DEBUG ERROR determining image orientation: {str(e)}")
orientation = "unknown"
# Return the image path as JSON
response = {
'path': f"/images/{image_path}",
@@ -367,6 +413,7 @@ class ImageSwipeHandler(BaseHTTPRequestHandler):
self.end_headers()
self.wfile.write(json.dumps(response).encode())
except Exception as e:
print(f"FATAL ERROR in serve_random_image: {e}")
self.send_error(500, f"Error serving random image: {str(e)}")
def serve_resolutions(self):
@@ -627,7 +674,14 @@ class ImageSwipeHandler(BaseHTTPRequestHandler):
def run(server_class=HTTPServer, handler_class=ImageSwipeHandler, port=8000):
# Initialize the database
init_db()
# Ensure the 'images' directory exists
if not os.path.exists(IMAGE_DIR):
os.makedirs(IMAGE_DIR)
# Sync the image database on startup
sync_image_database()
server_address = ('', port)
httpd = server_class(server_address, handler_class)
print(f"Starting server on port {port}...")