Compare commits

5 Commits

Author SHA1 Message Date
Jose
808ff8896e add: ej1 whoosh 2026-03-18 19:55:21 +01:00
Jose
ed61afac69 Change: module structure to single file structure as professor requested for some reason... 2026-03-15 04:31:11 +01:00
Jose
6e4c05665e Merge branch 'dev' of git.miarma.net:/Gallardo7761/aii-monorepo into dev 2026-03-07 22:09:39 +01:00
Jose
a0297672b2 add: whoosh 2026-03-07 22:09:00 +01:00
Jose
bcab404758 Finished 2026-02-22 12:45:08 +01:00
52 changed files with 1825 additions and 1529 deletions

View File

@@ -1,3 +1,4 @@
{
"python.REPL.enableREPLSmartSend": false
"python.REPL.enableREPLSmartSend": false,
"python-envs.defaultEnvManager": "ms-python.python:venv"
}

View File

@@ -1,5 +0,0 @@
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context

View File

@@ -1,6 +0,0 @@
from pathlib import Path
URL = "https://www.vinissimus.com/es/vinos/tinto/?cursor="
DATA_DIR = Path(__file__).parent.parent / "data"
CSV_PATH = DATA_DIR / "books.csv"
DB_PATH = DATA_DIR / "books.bd"

View File

@@ -1,141 +0,0 @@
import sqlite3
from pathlib import Path
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_singleton(self, singleton_table):
try:
cursor = self.conn.execute(f"SELECT * FROM {singleton_table}")
return [row[0] for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def clear(self, table_name):
query = f"DELETE FROM {table_name};"
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error clearing table: ", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()

View File

@@ -1,110 +0,0 @@
from bs4 import BeautifulSoup
import re
from tkinter import Tk
from tkinter import messagebox
import urllib.request
from db import DBManager, DBAttr
from ui import WinesUI
from __ssl import init_ssl
from config import *
init_ssl()
dbm = DBManager(DB_PATH)
def create_tables():
wines_attr = [
DBAttr("name", "TEXT", "NOT NULL"),
DBAttr("price", "INTEGER", "NOT NULL"),
DBAttr("origin", "TEXT", "NOT NULL"),
DBAttr("cellar", "TEXT", "NOT NULL"),
DBAttr("type", "TEXT", "NOT NULL")
]
types_attr = [
DBAttr("type", "TEXT")
]
dbm.create_table("wines", wines_attr)
dbm.create_table("types", types_attr)
def extract_wines():
l = []
for i in range(0,3):
f = urllib.request.urlopen(URL+str(i*36))
doc = BeautifulSoup(f, "lxml")
page = doc.find_all("div", class_="product-list-item")
l.extend(page)
return l
def persist_wines(wines):
types = set()
for wine in wines:
details = wine.find("div",class_=["details"])
name = details.a.h2.string.strip()
price = list(wine.find("p",class_=["price"]).stripped_strings)[0]
origin = details.find("div",class_=["region"]).string.strip()
cellar = details.find("div", class_=["cellar-name"]).string.strip()
grapes = "".join(details.find("div",class_=["tags"]).stripped_strings)
for g in grapes.split("/"):
types.add(g.strip())
disc = wine.find("p",class_=["price"]).find_next_sibling("p",class_="dto")
if disc:
price = list(disc.stripped_strings)[0]
dbm.insert("wines", {"name": name, "price": float(price.replace(',', '.')), "origin": origin, "cellar": cellar, "type": grapes})
for type in types:
dbm.insert("types", {"type": type})
return dbm.count("wines"), dbm.count("types")
def main():
create_tables()
root = Tk()
ui = WinesUI(root)
def handle_action(action):
match(action):
case "cargar":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
dbm.clear("wines")
dbm.clear("types")
wines = extract_wines()
wines_count, types_count = persist_wines(wines)
ui.info(f"Hay {wines_count} vinos y {types_count} uvas.")
case "listar":
wines = dbm.get_all("wines")
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
case "buscar_denominacion":
origins = list({wine["origin"] for wine in dbm.get_all("wines")})
origins.sort()
def search_origin(origin):
wines = [wine for wine in dbm.get_all("wines") if wine["origin"] == origin]
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
ui.ask_spinbox("Buscar por denominación: ", origins, search_origin)
case "buscar_precio":
def search_price(price):
wines = [wine for wine in dbm.get_all("wines") if float(wine["price"]) <= float(price)]
wines.sort(key=lambda w: float(w["price"]))
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
ui.ask_text("Selecciona precio: ", search_price)
case "buscar_uva":
types = [t for t in dbm.get_singleton("types")]
types.sort()
def search_type(type):
wines = [wine for wine in dbm.get_all("wines") if type in wine["type"]]
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
ui.ask_spinbox("Selecciona tip de uva: ", types, search_type)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

View File

@@ -1,76 +0,0 @@
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
class WinesUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_command(label="Listar", command=lambda: self.callback("listar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Denominación", command=lambda: self.callback("buscar_denominacion"))
buscar_menu.add_command(label="Precio", command=lambda: self.callback("buscar_precio"))
buscar_menu.add_command(label="Uva", command=lambda: self.callback("buscar_uva"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)

View File

@@ -1,5 +0,0 @@
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context

View File

@@ -1,6 +0,0 @@
from pathlib import Path
BASE_URL = "https://www.elseptimoarte.net"
ESTRENOS_URL = BASE_URL + "/estrenos/2025/"
DATA_DIR = Path(__file__).parent.parent / "data"
DB_PATH = DATA_DIR / "movies.bd"

View File

@@ -1,141 +0,0 @@
import sqlite3
from pathlib import Path
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_singleton(self, singleton_table):
try:
cursor = self.conn.execute(f"SELECT * FROM {singleton_table}")
return [row[0] for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def clear(self, table_name):
query = f"DELETE FROM {table_name};"
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error clearing table: ", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()

View File

@@ -1,109 +0,0 @@
from bs4 import BeautifulSoup
import re
from tkinter import Tk
from tkinter import messagebox
import urllib.request
from datetime import datetime
from db import DBManager, DBAttr
from ui import WinesUI
from __ssl import init_ssl
from config import *
init_ssl()
dbm = DBManager(DB_PATH)
def create_tables():
movies_attr = [
DBAttr("title", "TEXT", "NOT NULL"),
DBAttr("original_title", "TEXT", "NOT NULL"),
DBAttr("country", "TEXT", "NOT NULL"),
DBAttr("date", "DATE", "NOT NULL"),
DBAttr("director", "TEXT", "NOT NULL"),
DBAttr("genres", "TEXT", "NOT NULL")
]
genres_attr = [
DBAttr("genre", "TEXT")
]
dbm.create_table("movies", movies_attr)
dbm.create_table("genres", genres_attr)
def persist_movies():
f = urllib.request.urlopen(ESTRENOS_URL)
bs = BeautifulSoup(f, "lxml")
list_items = bs.find("ul", class_="elements").find_all("li")
for li in list_items:
f = urllib.request.urlopen(BASE_URL+li.a['href'])
bs = BeautifulSoup(f, "lxml")
data = bs.find("main", class_="informativo").find("section",class_="highlight").div.dl
original_title = data.find("dt", string=lambda s: s and "Título original" in s).find_next_sibling("dd").get_text(strip=True)
country = "".join(data.find("dt", string=lambda s: s and "País" in s).find_next_sibling("dd").stripped_strings)
title = data.find("dt", string=lambda s: s and "Título" in s).find_next_sibling("dd").get_text(strip=True)
date = datetime.strptime(data.find("dt",string="Estreno en España").find_next_sibling("dd").string.strip(), '%d/%m/%Y')
genres_director = bs.find("div",id="datos_pelicula")
genres_str = genres_director.find("p", class_="categorias").get_text(strip=True)
genres_list = [g.strip() for g in genres_str.split(",") if g.strip()]
for g in genres_list:
existing = dbm.exists("genres", "genre", g)
if not existing:
dbm.insert("genres", {"genre": g})
director = "".join(genres_director.find("p",class_="director").stripped_strings)
dbm.insert("movies", {
"title": title,
"original_title": original_title,
"country": country,
"date": date,
"director": director,
"genres": genres_str
})
return dbm.count("movies"), dbm.count("genres")
def main():
create_tables()
root = Tk()
ui = WinesUI(root)
def handle_action(action):
match(action):
case "cargar":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
dbm.clear("movies")
dbm.clear("genres")
movies_count, genres_count = persist_movies()
ui.info(f"Hay {movies_count} películas y {genres_count} géneros")
case "listar":
movies = dbm.get_all("movies")
ui.show_list(movies, ["title", "original_title", "country", "date", "director", "genres"])
case "buscar_titulo":
def search_title(title):
movies = [movie for movie in dbm.get_all("movies") if title.lower() in movie["title"].lower()]
ui.show_list(movies, ["title", "country", "director"])
ui.ask_text("Buscar por titulo: ", search_title)
case "buscar_fecha":
def search_date(date):
d = datetime.strptime(date, "%d-%m-%Y")
movies = [movie for movie in dbm.get_all("movies")
if d < datetime.strptime(movie["date"], "%Y-%m-%d %H:%M:%S")]
ui.show_list(movies, ["title", "date"])
ui.ask_text("Buscar por fecha: ", search_date)
case "buscar_genero":
genres = [g for g in dbm.get_singleton("genres")]
genres.sort()
def search_genre(genre):
movies = [movie for movie in dbm.get_all("movies") if genre in movie["genres"]]
ui.show_list(movies, ["title", "date"])
ui.ask_spinbox("Selecciona género: ", genres, search_genre)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

View File

@@ -1,76 +0,0 @@
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
class WinesUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_command(label="Listar", command=lambda: self.callback("listar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Título", command=lambda: self.callback("buscar_titulo"))
buscar_menu.add_command(label="Fecha", command=lambda: self.callback("buscar_fecha"))
buscar_menu.add_command(label="Género", command=lambda: self.callback("buscar_genero"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)

View File

@@ -1,5 +0,0 @@
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context

View File

@@ -1,6 +0,0 @@
from pathlib import Path
BASE_URL = "https://www.recetasgratis.net"
RECIPES_URL = BASE_URL + "/Recetas-de-Aperitivos-tapas-listado_receta-1_1.html/"
DATA_DIR = Path(__file__).parent.parent / "data"
DB_PATH = DATA_DIR / "recetas.bd"

View File

@@ -1,141 +0,0 @@
import sqlite3
from pathlib import Path
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_singleton(self, singleton_table):
try:
cursor = self.conn.execute(f"SELECT * FROM {singleton_table}")
return [row[0] for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def clear(self, table_name):
query = f"DELETE FROM {table_name};"
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error clearing table: ", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()

View File

@@ -1,85 +0,0 @@
from bs4 import BeautifulSoup
import re
from tkinter import Tk
from tkinter import messagebox
import urllib.request
from datetime import datetime
from db import DBManager, DBAttr
from ui import WinesUI
from __ssl import init_ssl
from config import *
init_ssl()
dbm = DBManager(DB_PATH)
def create_tables():
movies_attr = [
DBAttr("title", "TEXT", "NOT NULL"),
DBAttr("original_title", "TEXT", "NOT NULL"),
DBAttr("country", "TEXT", "NOT NULL"),
DBAttr("date", "DATE", "NOT NULL"),
DBAttr("director", "TEXT", "NOT NULL"),
DBAttr("genres", "TEXT", "NOT NULL")
]
genres_attr = [
DBAttr("genre", "TEXT")
]
dbm.create_table("movies", movies_attr)
dbm.create_table("genres", genres_attr)
def parse_duration(duration):
duration.strip()
res = 0
if duration[-1] == "h":
duration.replace("h","")
res = int(duration) * 60
elif "h" in duration:
duration.replace("h","")
duration.replace("m","")
res = int(duration[0]) + int(duration[1:])
else:
duration.replace("m","")
res = int(duration)
return res
def main():
create_tables()
root = Tk()
ui = WinesUI(root)
def handle_action(action):
match(action):
case "cargar":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
dbm.clear("movies")
dbm.clear("genres")
movies_count = persit_recetas()
ui.info(f"Hay {movies_count} recetas")
case "Recetas":
recetas = dbm.get_all("recetas")
ui.show_list(recetas, ["titulo", "dificultad", "comensales", "duracion", "autor", "fecha"])
case "Receta por autor":
def search_title(title):
movies = [movie for movie in dbm.get_all("recetas") if title.lower() in movie["titulo"].lower()]
ui.show_list(recetas, ["titulo", "dificultad", "comensales", "duracion", "autor", "fecha"])
ui.ask_text("Buscar por autor: ", search_title)
case "Receta por fecha":
def search_date(date):
d = datetime.strptime(date, "%d-%m-%Y")
movies = [movie for movie in dbm.get_all("movies")
if d < datetime.strptime(movie["date"], "%Y-%m-%d %H:%M:%S")]
ui.show_list(movies, ["title", "date"])
ui.ask_text("Buscar por fecha: ", search_date)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

View File

@@ -1,79 +0,0 @@
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
class WinesUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Listar
listar_menu = tk.Menu(self.menu, tearoff=0)
listar_menu.add_command(label= "Recetas", command = lambda: self.callback("listar_recetas"))
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Receta por autor", command=lambda: self.callback("buscar_autor"))
buscar_menu.add_command(label="Receta por fecha", command=lambda: self.callback("buscar_fecha"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)

View File

@@ -1,5 +0,0 @@
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context

View File

@@ -1,6 +0,0 @@
from pathlib import Path
BASE_URL = "https://recetas.elperiodico.com"
RECIPES_URL = BASE_URL + "/Recetas-de-Aperitivos-tapas-listado_receta-1_1.html"
DATA_DIR = Path(__file__).parent.parent / "data"
DB_PATH = DATA_DIR / "recipes.bd"

View File

@@ -1,141 +0,0 @@
import sqlite3
from pathlib import Path
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_singleton(self, singleton_table):
try:
cursor = self.conn.execute(f"SELECT * FROM {singleton_table}")
return [row[0] for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def clear(self, table_name):
query = f"DELETE FROM {table_name};"
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error clearing table: ", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()

View File

@@ -1,80 +0,0 @@
from bs4 import BeautifulSoup
import re
from tkinter import Tk
from tkinter import messagebox
import urllib.request
from datetime import datetime
import locale
from db import DBManager, DBAttr
#from ui import RecipesUI
from __ssl import init_ssl
from config import *
init_ssl()
locale.setlocale(locale.LC_TIME, "es_ES.UTF-8")
dbm = DBManager(DB_PATH)
def create_tables():
recipes_attr = [
DBAttr("title", "TEXT", "NOT NULL"),
DBAttr("difficulty", "TEXT", "DEFAULT NULL"),
DBAttr("units", "INTEGER", "DEFAULT NULL"),
DBAttr("duration", "INTEGER", "DEFAULT NULL"),
DBAttr("author", "TEXT", "NOT NULL"),
DBAttr("updated_at", "DATE", "NOT NULL")
]
dbm.create_table("recipes", recipes_attr)
def persist_recipes():
f = urllib.request.urlopen(RECIPES_URL)
bs = BeautifulSoup(f, "lxml")
results = bs.find_all("div", attrs={"data-js-selector": "resultado"})
for div in results:
print(div)
title_a = div.a
title = title_a.string.strip()
info_div = div.find("div", class_="info_snippet")
difficulty = info_div.find("span").get_text(strip=True) if info_div and info_div.find("span") else None
properties = div.find("div", class_="properties")
duration = properties.find("span", class_=["property", "duracion"]).string.strip() if properties and properties.find("span", class_=["property", "duracion"]) else None
units = properties.find("span", class_=["property", "unidades"]).string.strip() if properties and properties.find("span", class_=["property", "unidades"]) else None
details_link = title_a["href"]
f2 = urllib.request.urlopen(details_link)
bs2 = BeautifulSoup(f2, "lxml")
details = bs2.find("div", class_="autor").find("div", class_="nombre_autor")
author = details.find("a").string
date_str = details.find("span").string
updated_at = datetime.strptime(date_str, "%d %B %Y")
dbm.insert("recipes", {
"title": title,
"difficulty": difficulty,
"units": units,
"duration": duration,
"author": author,
"updated_at": updated_at
})
return dbm.count("recipes")
def main():
create_tables()
recipes_count = persist_recipes()
print(recipes_count)
#root = Tk()
#ui = RecipesUI(root)
# def handle_action(action):
#ui.callback = handle_action
#root.mainloop()
#dbm.close()
print(dbm.get_all("recipes"))
if __name__ == "__main__":
main()

View File

@@ -1,124 +0,0 @@
import sqlite3
from pathlib import Path
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()

View File

@@ -1,23 +0,0 @@
import csv
class FileReader:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, delimiter=";"):
self.delimiter = delimiter
def read(self, file):
results = []
try:
with open(file, encoding="utf-8-sig") as f:
reader = csv.DictReader(f, delimiter=self.delimiter)
for row in reader:
results.append(dict(row))
except Exception as e:
print(f"Error leyendo archivo: {e}")
return results

View File

@@ -1,77 +0,0 @@
from pathlib import Path
from tkinter import Tk
from files import FileReader
from db import DBManager, DBAttr
from ui import BooksUI
DATA_DIR = Path(__file__).parent.parent / "data"
CSV_PATH = DATA_DIR / "books.csv"
DB_PATH = DATA_DIR / "books.bd"
dbm = DBManager(DB_PATH)
fr = FileReader()
def create_tables():
book_attrs = [
DBAttr("isbn", "INTEGER", "PRIMARY KEY"),
DBAttr("title", "TEXT", "NOT NULL"),
DBAttr("author", "TEXT"),
DBAttr("year", "DATE"),
DBAttr("publisher", "TEXT")
]
dbm.create_table("books", book_attrs)
def main():
create_tables()
root = Tk()
ui = BooksUI(root)
def handle_action(action):
match(action):
case "cargar":
books = fr.read(CSV_PATH)
count = 0
for book in books:
book["isbn"] = int(book["isbn"])
if not dbm.exists("books", "isbn", book["isbn"]):
dbm.insert("books", book)
count += 1
ui.info(f"{count} libros almacenados.")
case "listar_todo":
books = dbm.get_all("books")
ui.show_list(books, ["isbn", "title", "author", "year"])
case "listar_ordenado":
def sort(attr):
books = dbm.get_all("books")
def key_fn(x):
v = x[attr]
if isinstance(v, int):
return v
elif isinstance(v, str) and v.isdigit():
return int(v)
else:
return float('inf')
books.sort(key=key_fn)
ui.show_list(books, ["isbn", "title", "author", "year"])
ui.ask_radiobutton("Ordenar por: ", ["isbn", "year"], sort)
case "buscar_titulo":
def search_title(title):
books = [book for book in dbm.get_all("books") if title.lower() in book["title"].lower()]
ui.show_list(books, ["isbn", "title", "author", "year"])
ui.ask_text("Buscar por título: ", search_title)
case "buscar_editorial":
publishers = list({book["publisher"] for book in dbm.get_all("books")})
publishers.sort()
def search_publisher(publisher):
books = [book for book in dbm.get_all("books") if book["publisher"] == publisher]
ui.show_list(books, ["title", "author", "publisher"])
ui.ask_spinbox("Selecciona editorial: ", publishers, search_publisher)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

View File

@@ -1,80 +0,0 @@
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
class BooksUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Listar
listar_menu = tk.Menu(self.menu, tearoff=0)
listar_menu.add_command(label="Completo", command=lambda: self.callback("listar_completo"))
listar_menu.add_command(label="Ordenado", command=lambda: self.callback("listar_ordenado"))
self.menu.add_cascade(label="Listar", menu=listar_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Título", command=lambda: self.callback("buscar_titulo"))
buscar_menu.add_command(label="Editorial", command=lambda: self.callback("buscar_editorial"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, books, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for book in books:
row = " | ".join(str(book[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)

View File

@@ -0,0 +1,8 @@
unoarrobagmail.com
Antonio Garcia
dosarrobagmail.com
Pedro Guerra
tresarrobagmail.com
Ana Montero
cuatroarrobagmail.com
Luis Pontes

View File

@@ -0,0 +1,10 @@
unoarrobagmail.com
dosarrobagmail.com tresarrobagmail.com
20101015
Contrato de compraventa con la constructora
Estimados socios:
ya hemos firmado el contrato de compraventa con el cliente preferencial.
Espero noticias vuestras.
Un saludo,

View File

@@ -0,0 +1,10 @@
dosarrobagmail.com
unoarrobagmail.com
20100410
Retraso en la firma del Contrato
Estimados Antonio:
agradezco mucho tus buenas noticias, aunque me temo que el documento que debe adjuntarse al contrato se va a retrasar
unos dias.
Un saludo,

View File

@@ -0,0 +1,10 @@
tresarrobagmail.com
unoarrobagmail.com dosarrobagmail.com
20140225
Transferencia realizada
Estimados socios:
aunque el contrato no este legalizado aun, me he permitido hacer una transferencia por
la mitad del importe al contratista.
Un saludo,

View File

@@ -0,0 +1,8 @@
unoarrobagmail.com
tresarrobagmail.com dosarrobagmail.com
20110114
Lo comunicare al cliente
Estimados socios:
muchas gracias por las gestiones. se lo comunicare al cliente hoy mismo.
Un saludo,

View File

@@ -0,0 +1,9 @@
unoarrobagmail.com
cuatroarrobagmail.com
20130912
Contrato y Transferencia
Estimado Luis:
ya hemos realizado una transferencia a su cuenta por el importe establecido inicialmente.
Un saludo,

View File

@@ -0,0 +1,6 @@
cuatroarrobagmail.com
unoarrobagmail.com
20131105
Gracias
Un saludo,

View File

@@ -0,0 +1,193 @@
import locale
import re
import urllib.request
from datetime import datetime
from pathlib import Path
import tkinter as tk
from tkinter import messagebox, ttk
from tkinter import Tk
from tkinter.scrolledtext import ScrolledText
import shutil, re, os
from whoosh.index import create_in,open_dir
from whoosh.fields import Schema, TEXT, DATETIME, KEYWORD, ID, NUMERIC
from whoosh.qparser import QueryParser
from whoosh import index, qparser, query
DATA_DIR = Path(__file__).parent / "data"
CONTACTS_DIR = DATA_DIR / "contacts"
EMAILS_DIR = DATA_DIR / "emails"
INDEX_DIR = Path(__file__).parent / "index"
CONTACTS = {}
def create_index():
if not os.path.exists(INDEX_DIR):
os.mkdir(INDEX_DIR)
if not index.exists_in(INDEX_DIR, indexname="EmailIndex"):
schema = Schema(sender=TEXT(stored=True),
receiver=KEYWORD(stored=True),
date=DATETIME(stored=True),
subject=TEXT(stored=True),
body=TEXT(stored=True,phrase=False),
file_name=ID(stored=True))
idx = create_in(INDEX_DIR, schema=schema, indexname="EmailIndex")
print(f"Created index: {idx.indexname}")
else:
print(f"An index already exists")
def add_to_index(writer, path, file_name):
try:
f = open(path, "r")
sender = f.readline().strip()
receiver = f.readline().strip()
date_raw = f.readline().strip()
date = datetime.strptime(date_raw, '%Y%m%d')
subject = f.readline().strip()
body = f.read()
f.close()
writer.add_document(
sender=sender,
receiver=receiver,
date=date,
subject=subject,
body=body,
file_name=file_name
)
except:
messagebox.showerror(f"[ERR] adding {path}/{file_name}")
def index_emails(delete = False):
if delete:
shutil.rmtree(INDEX_DIR)
os.mkdir(INDEX_DIR)
create_index()
idx = index.open_dir(INDEX_DIR, "EmailIndex")
writer = idx.writer()
count = 0
for f in os.listdir(EMAILS_DIR):
if not os.path.isdir(EMAILS_DIR / f):
add_to_index(writer, EMAILS_DIR / f, f)
count += 1
writer.commit()
return count
def create_contacts():
try:
f = open(CONTACTS_DIR / "agenda.txt", "r")
email = f.readline()
while email:
name = f.readline()
CONTACTS[email.strip()] = name.strip()
email = f.readline()
except:
messagebox.showerror(f"[ERR] creating contacts list")
def load(delete = False):
create_contacts()
return index_emails(delete)
class EmailsUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("load"))
datos_menu.add_command(label="Listar", command=lambda: self.callback("list"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Cuerpo o Asunto", command=lambda: self.callback("search_body_or_subject"))
buscar_menu.add_command(label="Fecha", command=lambda: self.callback("search_date"))
buscar_menu.add_command(label="Spam", command=lambda: self.callback("search_spam"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item.get(field, "Unknown")) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)
def main():
locale.setlocale(locale.LC_TIME, "es_ES.UTF-8")
create_index()
root = Tk()
ui = EmailsUI(root)
def handle_action(action):
match(action):
case "load":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
recipes_count = load(True)
ui.info(f"Se han indexado {recipes_count} emails")
case "list":
ix = open_dir(INDEX_DIR, "EmailIndex")
with ix.searcher() as searcher:
emails = searcher.search(query.Every(), limit=None)
print(emails)
ui.show_list(emails, ["sender", "receiver", "name", "subject", "body"])
# buscar con queries y tal...
ui.callback = handle_action
root.mainloop()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,247 @@
import locale
import re
import urllib.request
from datetime import datetime
from pathlib import Path
import tkinter as tk
from tkinter import messagebox, ttk
from tkinter import Tk
from tkinter.scrolledtext import ScrolledText
import shutil, re, os
from bs4 import BeautifulSoup
from whoosh.index import create_in,open_dir
from whoosh.fields import Schema, TEXT, DATETIME, KEYWORD, ID, NUMERIC
from whoosh.qparser import QueryParser
from whoosh import index, qparser, query
BASE_URL = "https://recetas.elperiodico.com"
RECIPES_URL = BASE_URL + "/Recetas-de-Aperitivos-tapas-listado_receta-1_1.html"
DATA_DIR = Path(__file__).parent / "index"
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context
def create_index():
if not os.path.exists(DATA_DIR):
os.mkdir(DATA_DIR)
if not index.exists_in(DATA_DIR, indexname="RecipesIndex"):
schema = Schema(
title=TEXT(stored=True),
difficulty=TEXT(stored=True),
duration=TEXT(stored=True),
units=NUMERIC(stored=True, numtype=int),
author=ID(stored=True),
updated_at=DATETIME(stored=True),
features=KEYWORD(stored=True, commas=True),
intro=TEXT(stored=True)
)
idx = create_in(DATA_DIR, schema=schema, indexname="RecipesIndex")
print(f"Created index: {idx.indexname}")
else:
print(f"An index already exists")
def parse_duration(duration):
if not duration:
return None
duration = duration.strip().lower()
hours = 0
minutes = 0
h_match = re.search(r"(\d+)h", duration)
m_match = re.search(r"(\d+)m", duration)
if h_match:
hours = int(h_match.group(1))
if m_match:
minutes = int(m_match.group(1))
return hours * 60 + minutes
def parse_duration_inverse(minutes):
if minutes is None:
return None
m = minutes % 60
h = (minutes - m) // 60
return f"{h}h {m}m" if h != 0 else f"{m}m"
def persist_recipes():
idx = index.open_dir(DATA_DIR, "RecipesIndex")
writer = idx.writer()
count = 0
f = urllib.request.urlopen(RECIPES_URL)
bs = BeautifulSoup(f, "lxml")
results = bs.find_all("div", attrs={"data-js-selector": "resultado"})
for div in results:
title_a = div.a
title = div.a.string.strip()
info_div = div.find("div", class_="info_snippet")
difficulty = info_div.find("span").get_text(strip=True) if info_div and info_div.find("span") else "Unknown"
intro = div.find("div", class_="intro").get_text()
properties = div.find("div", class_="properties")
duration = properties.find("span", class_="duracion").string.strip() if properties and properties.find("span", class_="duracion") else "Unknown"
units = int(properties.find("span", class_="unidades").string.strip()) if properties and properties.find("span", class_="unidades") else -1
details_link = title_a["href"]
f2 = urllib.request.urlopen(details_link)
bs2 = BeautifulSoup(f2, "lxml")
details = bs2.find("div", class_="autor").find("div", class_="nombre_autor")
author = details.find("a").string
date_str = details.find("span").string.replace("Actualizado: ", "")
updated_at = datetime.strptime(date_str, "%d %B %Y")
features = bs2.find("div", class_=["properties", "inline"]).get_text(strip=True).replace("Características adicionales:", "") if bs2.find("div", class_=["properties", "inline"]) else "Unknown"
writer.add_document(
title=title,
difficulty=difficulty,
duration=duration,
units=units,
author=author,
updated_at=updated_at,
features=features,
intro=intro
)
count += 1
writer.commit()
return count
class RecipesUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("load"))
datos_menu.add_command(label="Listar", command=lambda: self.callback("list_recipes"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Título o Introducción", command=lambda: self.callback("search_title_or_intro"))
buscar_menu.add_command(label="Fecha", command=lambda: self.callback("search_updated_at"))
buscar_menu.add_command(label="Características y Título", command=lambda: self.callback("search_features_and_title"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item.get(field, "Unknown")) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)
def main():
init_ssl()
locale.setlocale(locale.LC_TIME, "es_ES.UTF-8")
create_index()
root = Tk()
ui = RecipesUI(root)
def handle_action(action):
match(action):
case "load":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
recipes_count = persist_recipes()
ui.info(f"Se han indexado {recipes_count} recetas")
case "list_recipes":
ix = open_dir(DATA_DIR, "RecipesIndex")
with ix.searcher() as searcher:
recipes = searcher.search(query.Every(), limit=None)
clear = []
for r in recipes:
d = dict(r)
clear.append(d)
print(clear)
ui.show_list(clear, ["title", "difficulty", "units", "duration"])
# case "search_title_or_intro":
# def search_author(author):
# recipes = [recipe for recipe in dbm.get_all("recipes") if author.lower() in recipe["author"].lower()]
# for r in recipes:
# r["units"] = str(r["units"]) + " personas" if r["units"] is not None else "Unknown personas"
# r["duration"] = parse_duration_inverse(r["duration"])
# ui.show_list(recipes, ["title", "difficulty", "units", "duration", "author"])
# ui.ask_text("Buscar por autor: ", search_author)
# case "search_updated_at":
# def search_date(date):
# d = datetime.strptime(date, "%d/%m/%Y")
# recipes = [recipe for recipe in dbm.get_all("recipes")
# if d > datetime.strptime(recipe["updated_at"], "%Y-%m-%d %H:%M:%S")]
# for r in recipes:
# r["units"] = str(r["units"]) + " personas" if r["units"] is not None else "Unknown personas"
# r["duration"] = parse_duration_inverse(r["duration"])
# ui.show_list(recipes, ["title", "difficulty", "units", "duration", "updated_at"])
# ui.ask_text("Buscar por fecha: ", search_date)
# case "search_features_and_title":
# def search_author(author):
# recipes = [recipe for recipe in dbm.get_all("recipes") if author.lower() in recipe["author"].lower()]
# for r in recipes:
# r["units"] = str(r["units"]) + " personas" if r["units"] is not None else "Unknown personas"
# r["duration"] = parse_duration_inverse(r["duration"])
# ui.show_list(recipes, ["title", "difficulty", "units", "duration", "author"])
# ui.ask_text("Buscar por autor: ", search_author)
ui.callback = handle_action
root.mainloop()
if __name__ == "__main__":
main()

View File

@@ -1,6 +1,7 @@
import requests
import re
# --- CONSTANTS ------------------------------------------
RSS_URL = "https://www.abc.es/rss/2.0/espana/andalucia/"
ITEM_PATTERN = r"<item>(.*?)</item>"
MONTHS = {
@@ -9,6 +10,7 @@ MONTHS = {
"Sep": "09", "Oct": "10", "Nov": "11", "Dec": "12",
}
# --- MAIN PROGRAM FUNCTIONS -----------------------------
def get_tag(text, tag):
m = re.search(rf"<{tag}>(.*?)</{tag}>", text, re.DOTALL)
return m.group(1).strip() if m else None

View File

@@ -0,0 +1,300 @@
from pathlib import Path
from tkinter import Tk
import sqlite3
import csv
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
# --- CONSTANTS ------------------------------------------
DATA_DIR = Path(__file__).parent / "data"
CSV_PATH = DATA_DIR / "books.csv"
DB_PATH = DATA_DIR / "books.bd"
# --- HELPER CLASSES -------------------------------------
class FileReader:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, delimiter=";"):
self.delimiter = delimiter
def read(self, file):
results = []
try:
with open(file, encoding="utf-8-sig") as f:
reader = csv.DictReader(f, delimiter=self.delimiter)
for row in reader:
results.append(dict(row))
except Exception as e:
print(f"Error leyendo archivo: {e}")
return results
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()
class BooksUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Listar
listar_menu = tk.Menu(self.menu, tearoff=0)
listar_menu.add_command(label="Completo", command=lambda: self.callback("listar_completo"))
listar_menu.add_command(label="Ordenado", command=lambda: self.callback("listar_ordenado"))
self.menu.add_cascade(label="Listar", menu=listar_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Título", command=lambda: self.callback("buscar_titulo"))
buscar_menu.add_command(label="Editorial", command=lambda: self.callback("buscar_editorial"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, books, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for book in books:
row = " | ".join(str(book[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)
# --- MAIN PROGRAM FUNCTIONS -----------------------------
dbm = DBManager(DB_PATH)
fr = FileReader()
def create_tables():
book_attrs = [
DBAttr("isbn", "INTEGER", "PRIMARY KEY"),
DBAttr("title", "TEXT", "NOT NULL"),
DBAttr("author", "TEXT"),
DBAttr("year", "DATE"),
DBAttr("publisher", "TEXT")
]
dbm.create_table("books", book_attrs)
def main():
create_tables()
root = Tk()
ui = BooksUI(root)
def handle_action(action):
match(action):
case "cargar":
books = fr.read(CSV_PATH)
count = 0
for book in books:
book["isbn"] = int(book["isbn"])
if not dbm.exists("books", "isbn", book["isbn"]):
dbm.insert("books", book)
count += 1
ui.info(f"{count} libros almacenados.")
case "listar_todo":
books = dbm.get_all("books")
ui.show_list(books, ["isbn", "title", "author", "year"])
case "listar_ordenado":
def sort(attr):
books = dbm.get_all("books")
def key_fn(x):
v = x[attr]
if isinstance(v, int):
return v
elif isinstance(v, str) and v.isdigit():
return int(v)
else:
return float('inf')
books.sort(key=key_fn)
ui.show_list(books, ["isbn", "title", "author", "year"])
ui.ask_radiobutton("Ordenar por: ", ["isbn", "year"], sort)
case "buscar_titulo":
def search_title(title):
books = [book for book in dbm.get_all("books") if title.lower() in book["title"].lower()]
ui.show_list(books, ["isbn", "title", "author", "year"])
ui.ask_text("Buscar por título: ", search_title)
case "buscar_editorial":
publishers = list({book["publisher"] for book in dbm.get_all("books")})
publishers.sort()
def search_publisher(publisher):
books = [book for book in dbm.get_all("books") if book["publisher"] == publisher]
ui.show_list(books, ["title", "author", "publisher"])
ui.ask_spinbox("Selecciona editorial: ", publishers, search_publisher)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,332 @@
from bs4 import BeautifulSoup
import re
from tkinter import Tk
from tkinter import messagebox
import urllib.request
from pathlib import Path
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
import sqlite3
# --- CONSTANTS ------------------------------------------
URL = "https://www.vinissimus.com/es/vinos/tinto/?cursor="
DATA_DIR = Path(__file__).parent / "data"
CSV_PATH = DATA_DIR / "books.csv"
DB_PATH = DATA_DIR / "books.bd"
# --- HELPER CLASSES -------------------------------------
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_singleton(self, singleton_table):
try:
cursor = self.conn.execute(f"SELECT * FROM {singleton_table}")
return [row[0] for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def clear(self, table_name):
query = f"DELETE FROM {table_name};"
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error clearing table: ", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()
class WinesUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_command(label="Listar", command=lambda: self.callback("listar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Denominación", command=lambda: self.callback("buscar_denominacion"))
buscar_menu.add_command(label="Precio", command=lambda: self.callback("buscar_precio"))
buscar_menu.add_command(label="Uva", command=lambda: self.callback("buscar_uva"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)
# --- MAIN PROGRAM FUNCTIONS -----------------------------
dbm = DBManager(DB_PATH)
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context
def create_tables():
wines_attr = [
DBAttr("name", "TEXT", "NOT NULL"),
DBAttr("price", "INTEGER", "NOT NULL"),
DBAttr("origin", "TEXT", "NOT NULL"),
DBAttr("cellar", "TEXT", "NOT NULL"),
DBAttr("type", "TEXT", "NOT NULL")
]
types_attr = [
DBAttr("type", "TEXT")
]
dbm.create_table("wines", wines_attr)
dbm.create_table("types", types_attr)
def extract_wines():
l = []
for i in range(0,3):
f = urllib.request.urlopen(URL+str(i*36))
doc = BeautifulSoup(f, "lxml")
page = doc.find_all("div", class_="product-list-item")
l.extend(page)
return l
def persist_wines(wines):
types = set()
for wine in wines:
details = wine.find("div",class_=["details"])
name = details.a.h2.string.strip()
price = list(wine.find("p",class_=["price"]).stripped_strings)[0]
origin = details.find("div",class_=["region"]).string.strip()
cellar = details.find("div", class_=["cellar-name"]).string.strip()
grapes = "".join(details.find("div",class_=["tags"]).stripped_strings)
for g in grapes.split("/"):
types.add(g.strip())
disc = wine.find("p",class_=["price"]).find_next_sibling("p",class_="dto")
if disc:
price = list(disc.stripped_strings)[0]
dbm.insert("wines", {"name": name, "price": float(price.replace(',', '.')), "origin": origin, "cellar": cellar, "type": grapes})
for type in types:
dbm.insert("types", {"type": type})
return dbm.count("wines"), dbm.count("types")
def main():
create_tables()
root = Tk()
ui = WinesUI(root)
def handle_action(action):
match(action):
case "cargar":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
dbm.clear("wines")
dbm.clear("types")
wines = extract_wines()
wines_count, types_count = persist_wines(wines)
ui.info(f"Hay {wines_count} vinos y {types_count} uvas.")
case "listar":
wines = dbm.get_all("wines")
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
case "buscar_denominacion":
origins = list({wine["origin"] for wine in dbm.get_all("wines")})
origins.sort()
def search_origin(origin):
wines = [wine for wine in dbm.get_all("wines") if wine["origin"] == origin]
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
ui.ask_spinbox("Buscar por denominación: ", origins, search_origin)
case "buscar_precio":
def search_price(price):
wines = [wine for wine in dbm.get_all("wines") if float(wine["price"]) <= float(price)]
wines.sort(key=lambda w: float(w["price"]))
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
ui.ask_text("Selecciona precio: ", search_price)
case "buscar_uva":
types = [t for t in dbm.get_singleton("types")]
types.sort()
def search_type(type):
wines = [wine for wine in dbm.get_all("wines") if type in wine["type"]]
ui.show_list(wines, ["name", "price", "origin", "cellar", "type"])
ui.ask_spinbox("Selecciona tip de uva: ", types, search_type)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,330 @@
from bs4 import BeautifulSoup
import re
import urllib.request
from datetime import datetime
from pathlib import Path
import sqlite3
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
from tkinter import Tk
# --- CONSTANTS ------------------------------------------
BASE_URL = "https://www.elseptimoarte.net"
ESTRENOS_URL = BASE_URL + "/estrenos/2025/"
DATA_DIR = Path(__file__).parent / "data"
DB_PATH = DATA_DIR / "movies.bd"
# --- HELPER CLASSES -------------------------------------
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_singleton(self, singleton_table):
try:
cursor = self.conn.execute(f"SELECT * FROM {singleton_table}")
return [row[0] for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def clear(self, table_name):
query = f"DELETE FROM {table_name};"
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error clearing table: ", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()
class WinesUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_command(label="Listar", command=lambda: self.callback("listar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Título", command=lambda: self.callback("buscar_titulo"))
buscar_menu.add_command(label="Fecha", command=lambda: self.callback("buscar_fecha"))
buscar_menu.add_command(label="Género", command=lambda: self.callback("buscar_genero"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)
# --- MAIN PROGRAM FUNCTIONS -----------------------------
dbm = DBManager(DB_PATH)
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context
def create_tables():
movies_attr = [
DBAttr("title", "TEXT", "NOT NULL"),
DBAttr("original_title", "TEXT", "NOT NULL"),
DBAttr("country", "TEXT", "NOT NULL"),
DBAttr("date", "DATE", "NOT NULL"),
DBAttr("director", "TEXT", "NOT NULL"),
DBAttr("genres", "TEXT", "NOT NULL")
]
genres_attr = [
DBAttr("genre", "TEXT")
]
dbm.create_table("movies", movies_attr)
dbm.create_table("genres", genres_attr)
def persist_movies():
f = urllib.request.urlopen(ESTRENOS_URL)
bs = BeautifulSoup(f, "lxml")
list_items = bs.find("ul", class_="elements").find_all("li")
for li in list_items:
f = urllib.request.urlopen(BASE_URL+li.a['href'])
bs = BeautifulSoup(f, "lxml")
data = bs.find("main", class_="informativo").find("section",class_="highlight").div.dl
original_title = data.find("dt", string=lambda s: s and "Título original" in s).find_next_sibling("dd").get_text(strip=True)
country = "".join(data.find("dt", string=lambda s: s and "País" in s).find_next_sibling("dd").stripped_strings)
title = data.find("dt", string=lambda s: s and "Título" in s).find_next_sibling("dd").get_text(strip=True)
date = datetime.strptime(data.find("dt",string="Estreno en España").find_next_sibling("dd").string.strip(), '%d/%m/%Y')
genres_director = bs.find("div",id="datos_pelicula")
genres_str = genres_director.find("p", class_="categorias").get_text(strip=True)
genres_list = [g.strip() for g in genres_str.split(",") if g.strip()]
for g in genres_list:
existing = dbm.exists("genres", "genre", g)
if not existing:
dbm.insert("genres", {"genre": g})
director = "".join(genres_director.find("p",class_="director").stripped_strings)
dbm.insert("movies", {
"title": title,
"original_title": original_title,
"country": country,
"date": date,
"director": director,
"genres": genres_str
})
return dbm.count("movies"), dbm.count("genres")
def main():
create_tables()
root = Tk()
ui = WinesUI(root)
def handle_action(action):
match(action):
case "cargar":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
dbm.clear("movies")
dbm.clear("genres")
movies_count, genres_count = persist_movies()
ui.info(f"Hay {movies_count} películas y {genres_count} géneros")
case "listar":
movies = dbm.get_all("movies")
ui.show_list(movies, ["title", "original_title", "country", "date", "director", "genres"])
case "buscar_titulo":
def search_title(title):
movies = [movie for movie in dbm.get_all("movies") if title.lower() in movie["title"].lower()]
ui.show_list(movies, ["title", "country", "director"])
ui.ask_text("Buscar por titulo: ", search_title)
case "buscar_fecha":
def search_date(date):
d = datetime.strptime(date, "%d-%m-%Y")
movies = [movie for movie in dbm.get_all("movies")
if d < datetime.strptime(movie["date"], "%Y-%m-%d %H:%M:%S")]
ui.show_list(movies, ["title", "date"])
ui.ask_text("Buscar por fecha: ", search_date)
case "buscar_genero":
genres = [g for g in dbm.get_singleton("genres")]
genres.sort()
def search_genre(genre):
movies = [movie for movie in dbm.get_all("movies") if genre in movie["genres"]]
ui.show_list(movies, ["title", "date"])
ui.ask_spinbox("Selecciona género: ", genres, search_genre)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

Binary file not shown.

View File

@@ -0,0 +1,356 @@
from bs4 import BeautifulSoup
import re
import urllib.request
from datetime import datetime
import locale
from pathlib import Path
from tkinter import Tk
import tkinter as tk
from tkinter import ttk, messagebox
from tkinter.scrolledtext import ScrolledText
import sqlite3
# --- CONSTANTS ------------------------------------------
BASE_URL = "https://recetas.elperiodico.com"
RECIPES_URL = BASE_URL + "/Recetas-de-Aperitivos-tapas-listado_receta-1_1.html"
DATA_DIR = Path(__file__).parent / "data"
DB_PATH = DATA_DIR / "recipes.bd"
# --- HELPER CLASSES -------------------------------------
class DBAttr:
def __init__(self, name, type_, modifier=""):
self.name = name
self.type_ = type_
self.modifier = modifier
def sql(self):
parts = [self.name, self.type_]
if self.modifier:
parts.append(self.modifier)
return " ".join(parts)
class DBManager:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, path):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.row_factory = sqlite3.Row
def create_table(self, table_name, attributes: list[DBAttr]):
columns_sql = ",\n ".join(attr.sql() for attr in attributes)
query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_sql}
);
"""
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error creating table:", e)
def get_all(self, table_name):
try:
cursor = self.conn.execute(f"SELECT * FROM {table_name};")
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_singleton(self, singleton_table):
try:
cursor = self.conn.execute(f"SELECT * FROM {singleton_table}")
return [row[0] for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def get_by(self, table_name, column, value):
try:
query = f"SELECT * FROM {table_name} WHERE {column} = ?;"
cursor = self.conn.execute(query, (value,))
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
print("Error selecting:", e)
return []
def insert(self, table_name, data: dict):
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
values = tuple(data.values())
query = f"""
INSERT INTO {table_name} ({keys})
VALUES ({placeholders});
"""
try:
with self.conn:
self.conn.execute(query, values)
except Exception as e:
print("Error inserting:", e)
def update(self, table_name, data: dict, where_column, where_value):
set_clause = ", ".join(f"{key} = ?" for key in data.keys())
values = list(data.values())
values.append(where_value)
query = f"""
UPDATE {table_name}
SET {set_clause}
WHERE {where_column} = ?;
"""
try:
with self.conn:
self.conn.execute(query, tuple(values))
except Exception as e:
print("Error updating:", e)
def delete(self, table_name, where_column, where_value):
query = f"DELETE FROM {table_name} WHERE {where_column} = ?;"
try:
with self.conn:
self.conn.execute(query, (where_value,))
except Exception as e:
print("Error deleting:", e)
def clear(self, table_name):
query = f"DELETE FROM {table_name};"
try:
with self.conn:
self.conn.execute(query)
except Exception as e:
print("Error clearing table: ", e)
def exists(self, table_name, where_column, where_value):
query = f"SELECT 1 FROM {table_name} WHERE {where_column} = ? LIMIT 1;"
try:
cursor = self.conn.execute(query, (where_value,))
return cursor.fetchone() is not None
except Exception as e:
print("Error checking existence:", e)
return False
def count(self, table_name):
try:
cursor = self.conn.execute(f"SELECT COUNT(*) as total FROM {table_name};")
return cursor.fetchone()["total"]
except Exception as e:
print("Error counting:", e)
return 0
def close(self):
self.conn.close()
class RecipesUI():
def __init__(self, root, title = "AII"):
self.root = root
self.root.title(title)
self.root.geometry("900x600")
# Menu Principal
self.menu = tk.Menu(self.root)
self.root.config(menu=self.menu)
# Menu Datos
datos_menu = tk.Menu(self.menu, tearoff=0)
datos_menu.add_command(label="Cargar", command=lambda: self.callback("cargar"))
datos_menu.add_separator()
datos_menu.add_command(label="Salir", command=self.root.quit)
self.menu.add_cascade(label="Datos", menu=datos_menu)
# Menu Listar
listar_menu = tk.Menu(self.menu, tearoff=0)
listar_menu.add_command(label= "Recetas", command = lambda: self.callback("listar_recetas"))
self.menu.add_cascade(label="Listar", menu=listar_menu)
# Menu Buscar
buscar_menu = tk.Menu(self.menu, tearoff=0)
buscar_menu.add_command(label="Receta por autor", command=lambda: self.callback("buscar_autor"))
buscar_menu.add_command(label="Receta por fecha", command=lambda: self.callback("buscar_fecha"))
self.menu.add_cascade(label="Buscar", menu=buscar_menu)
# Callback externo desde el punto de entrada
self.callback = None
def show_list(self, items, fields, title="Listado"):
mw = tk.Toplevel(self.root)
mw.title(title)
listbox = tk.Listbox(mw, width=80, height=20)
listbox.pack(side="left", fill="both", expand=True)
scrollbar = tk.Scrollbar(mw)
scrollbar.pack(side="right", fill="y")
listbox.config(yscrollcommand=scrollbar.set)
scrollbar.config(command=listbox.yview)
for item in items:
row = " | ".join(str(item[field]) for field in fields)
listbox.insert("end", row)
def ask_text(self, label, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
entry = ttk.Entry(mw)
entry.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(entry.get()), mw.destroy()]).pack(pady=10)
def ask_spinbox(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
spinbox = ttk.Spinbox(mw, values=options, state="readonly", width=40)
spinbox.pack(pady=5)
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(spinbox.get()), mw.destroy()]).pack(pady=10)
def ask_radiobutton(self, label, options, callback):
mw = tk.Toplevel(self.root)
mw.title(label)
tk.Label(mw, text=label).pack(pady=5)
sv = tk.StringVar(value=options[0])
for option in options:
tk.Radiobutton(mw, text=option, variable=sv, value=option).pack(anchor="w")
ttk.Button(mw, text="Aceptar", command=
lambda: [callback(sv.get()), mw.destroy()]).pack(pady=10)
def info(slef, message):
messagebox.showinfo("Información", message)
# --- MAIN PROGRAM FUNCTIONS -----------------------------
dbm = DBManager(DB_PATH)
def init_ssl():
import os, ssl
if (not os.environ.get('PYTHONHTTPSVERIFY', '') and
getattr(ssl, '_create_unverified_context', None)):
ssl._create_default_https_context = ssl._create_unverified_context
def create_tables():
recipes_attr = [
DBAttr("title", "TEXT", "NOT NULL"),
DBAttr("difficulty", "TEXT", "DEFAULT NULL"),
DBAttr("units", "INTEGER", "DEFAULT NULL"),
DBAttr("duration", "INTEGER", "DEFAULT NULL"),
DBAttr("author", "TEXT", "NOT NULL"),
DBAttr("updated_at", "DATE", "NOT NULL")
]
dbm.create_table("recipes", recipes_attr)
def parse_duration(duration):
if not duration:
return None
duration = duration.strip().lower()
hours = 0
minutes = 0
h_match = re.search(r"(\d+)h", duration)
m_match = re.search(r"(\d+)m", duration)
if h_match:
hours = int(h_match.group(1))
if m_match:
minutes = int(m_match.group(1))
return hours * 60 + minutes
def parse_duration_inverse(minutes):
if minutes is None:
return None
m = minutes % 60
h = (minutes - m) // 60
return f"{h}h {m}m" if h != 0 else f"{m}m"
def persist_recipes():
f = urllib.request.urlopen(RECIPES_URL)
bs = BeautifulSoup(f, "lxml")
results = bs.find_all("div", attrs={"data-js-selector": "resultado"})
for div in results:
title_a = div.a
title = title_a.string.strip()
info_div = div.find("div", class_="info_snippet")
difficulty = info_div.find("span").get_text(strip=True) if info_div and info_div.find("span") else None
properties = div.find("div", class_="properties")
duration = properties.find("span", class_="duracion").string.strip() if properties and properties.find("span", class_="duracion") else None
units = properties.find("span", class_="unidades").string.strip() if properties and properties.find("span", class_="unidades") else None
details_link = title_a["href"]
f2 = urllib.request.urlopen(details_link)
bs2 = BeautifulSoup(f2, "lxml")
details = bs2.find("div", class_="autor").find("div", class_="nombre_autor")
author = details.find("a").string
date_str = details.find("span").string.replace("Actualizado: ", "")
updated_at = datetime.strptime(date_str, "%d %B %Y")
dbm.insert("recipes", {
"title": title,
"difficulty": difficulty,
"units": units,
"duration": parse_duration(duration),
"author": author,
"updated_at": updated_at
})
return dbm.count("recipes")
def main():
locale.setlocale(locale.LC_TIME, "es_ES.UTF-8")
create_tables()
root = Tk()
ui = RecipesUI(root)
def handle_action(action):
match(action):
case "cargar":
resp = messagebox.askyesno(title="Cargar", message="Quieres cargar todos los datos de nuevo?")
if resp:
dbm.clear("recipes")
recipes_count = persist_recipes()
ui.info(f"Hay {recipes_count} recetas")
case "listar_recetas":
recipes = dbm.get_all("recipes")
for r in recipes:
r["units"] = str(r["units"]) + " personas" if r["units"] is not None else "Unknown personas"
r["duration"] = parse_duration_inverse(r["duration"])
ui.show_list(recipes, ["title", "difficulty", "units", "duration"])
case "buscar_autor":
def search_author(author):
recipes = [recipe for recipe in dbm.get_all("recipes") if author.lower() in recipe["author"].lower()]
for r in recipes:
r["units"] = str(r["units"]) + " personas" if r["units"] is not None else "Unknown personas"
r["duration"] = parse_duration_inverse(r["duration"])
ui.show_list(recipes, ["title", "difficulty", "units", "duration", "author"])
ui.ask_text("Buscar por autor: ", search_author)
case "buscar_fecha":
def search_date(date):
d = datetime.strptime(date, "%d/%m/%Y")
recipes = [recipe for recipe in dbm.get_all("recipes")
if d > datetime.strptime(recipe["updated_at"], "%Y-%m-%d %H:%M:%S")]
for r in recipes:
r["units"] = str(r["units"]) + " personas" if r["units"] is not None else "Unknown personas"
r["duration"] = parse_duration_inverse(r["duration"])
ui.show_list(recipes, ["title", "difficulty", "units", "duration", "updated_at"])
ui.ask_text("Buscar por fecha: ", search_date)
ui.callback = handle_action
root.mainloop()
dbm.close()
if __name__ == "__main__":
main()

View File

@@ -1,3 +1,4 @@
lxml
bs4
requests
requests
whoosh