-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
62 lines (51 loc) · 1.62 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import dotenv
import os
import sqlite3
import sys
from loguru import logger
import pandas as pd
dotenv.load_dotenv()
logger.remove(0)
logger.add(
"./activity.log",
format="{time}: {level} {message}",
colorize=True,
enqueue=True,
)
logger.add(
sys.stderr,
format="{time}: {level} {message}",
colorize=True,
enqueue=True,
)
def export_data(db_path: str, tables: list[str], xlsx_path: str) -> None:
try:
logger.info("Exporting data...")
# Connect to SQLite database
conn = sqlite3.connect(db_path)
# Create Excel writer object
with pd.ExcelWriter(xlsx_path) as writer:
# Export each table to a different sheet
for table in tables:
query = f"SELECT * FROM {table}"
df = pd.read_sql_query(query, conn)
df.to_excel(writer, sheet_name=table, index=False)
logger.info(f"Exported table {table} to sheet {table}")
logger.info(f"Successfully exported all tables to {xlsx_path}")
except Exception as e:
logger.error(f"Error: {str(e)}")
finally:
# Close the connection
conn.close()
# Usage
db_path = os.getenv("SQLITE_PATH", None)
tables = [x.strip() for x in os.getenv("TABLES", "").split(",") if x.strip()]
xlsx_path = os.getenv("OUTPUT_PATH", "output.xlsx")
if __name__ == "__main__":
if not db_path or not os.path.exists(db_path):
logger.error(f"Error: {db_path} does not exist.")
sys.exit(1)
if len(tables) == 0:
logger.error("Error: No tables specified.")
sys.exit(1)
export_data(db_path, tables, xlsx_path)