Compare commits

...

7 Commits

Author SHA1 Message Date
Jose
374a90d023 Add: exercise 2 BeautifulSoup 2026-02-16 03:54:05 +01:00
Jose
65b2322022 Add: exercise 1 BeautifulSoup 2026-02-16 02:40:41 +01:00
Jose
2dbbf387d6 Finished: basic exercises. Started: BS4 exercise 1. 2026-02-15 05:17:23 +01:00
Jose
a9effed3b0 Change: project dir structure 2026-02-11 18:24:46 +01:00
Jose
d4bc8b5d6d Add: ej1 and started ej2 2026-02-10 14:49:28 +01:00
Jose
9ee0cb6a87 Add: ej1 and started ej2 2026-02-10 02:33:10 +01:00
Jose
b94c178fb4 Initial commit 2026-02-10 00:31:30 +01:00
21 changed files with 4248 additions and 0 deletions

3
.vscode/settings.json vendored Normal file
View File

@@ -0,0 +1,3 @@
{
"python.REPL.enableREPLSmartSend": false
}

Binary file not shown.

View File

@@ -0,0 +1,5 @@
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

@@ -0,0 +1,6 @@
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

@@ -0,0 +1,141 @@
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

@@ -0,0 +1,110 @@
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

@@ -0,0 +1,76 @@
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)

Binary file not shown.

View File

@@ -0,0 +1,5 @@
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

@@ -0,0 +1,6 @@
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

@@ -0,0 +1,141 @@
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

@@ -0,0 +1,109 @@
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

@@ -0,0 +1,76 @@
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

@@ -0,0 +1,60 @@
import requests
import re
RSS_URL = "https://www.abc.es/rss/2.0/espana/andalucia/"
ITEM_PATTERN = r"<item>(.*?)</item>"
MONTHS = {
"Jan": "01", "Feb": "02", "Mar": "03", "Apr": "04",
"May": "05", "Jun": "06", "Jul": "07", "Aug": "08",
"Sep": "09", "Oct": "10", "Nov": "11", "Dec": "12",
}
def get_tag(text, tag):
m = re.search(rf"<{tag}>(.*?)</{tag}>", text, re.DOTALL)
return m.group(1).strip() if m else None
def format_date(raw):
if not raw:
return None
m = re.search(r"\w{3}, (\d{2}) (\w{3}) (\d{4})", raw)
if not m:
return None
d, mon, y = m.groups()
return f"{d}/{MONTHS[mon]}/{y}"
def format_item(item):
return (
f"Título: {item['title']}\n"
f"Link: {item['link']}\n"
f"Fecha: {item['date']}\n"
)
def get_raw():
rss = requests.get(RSS_URL).text
return re.findall(ITEM_PATTERN, rss, re.DOTALL)
def get_parsed():
parsed = []
for item in get_raw():
parsed.append({
"title": get_tag(item, "title"),
"link": get_tag(item, "link"),
"date": format_date(get_tag(item, "pubDate")),
})
return parsed
def main():
user_month = input("Type a month (MM): ")
user_day = input("Type a day (DD): ")
d = f"{user_day}/{user_month}"
for item in get_parsed():
if item["date"] and item["date"].startswith(d):
print(format_item(item))
if __name__ == "__main__":
main()

Binary file not shown.

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,124 @@
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

@@ -0,0 +1,23 @@
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

@@ -0,0 +1,77 @@
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

@@ -0,0 +1,80 @@
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)

3
requirements.txt Normal file
View File

@@ -0,0 +1,3 @@
lxml
bs4
requests