-
Notifications
You must be signed in to change notification settings - Fork 1
/
ProcessNotion.py
154 lines (130 loc) · 5.25 KB
/
ProcessNotion.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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import os
import csv
import pandas as pd
from notion_client import *
import logging
import asyncio
from telethon import TelegramClient
from telethon.tl.types import InputMessagesFilterPhotos
class NotionDatabaseDW:
def __init__(self, notion_token, database_id):
self.notion = Client(auth=notion_token)
self.database_id = database_id
def load_image_to_page(self, page_id, image_url):
logging.info("Adding image to Notion page...")
block_properties = {
"object": "block",
"type": "embed",
"embed": {
"url": image_url
}
}
response = self.notion.blocks.children.append(
block_id=page_id,
children=[block_properties]
)
if response:
logging.info(f"Image {image_url} added to the page successfully!")
else:
logging.info(f"Failed to add image {image_url} to the page.")
def add_embed_to_page(self, page_id, source, width=None, height=None):
logging.info("Adding embed block to Notion page...")
# Create a new block for the embedded content
block_properties = {
"object": "block",
"type": "embed",
"embed": {
"url": source
}
}
# Add optional properties to the block
# if width:
# block_properties["embed"]["width"] = width
# if height:
# block_properties["embed"]["height"] = height
# Append the new block to the Notion page
try:
response = self.notion.blocks.children.append(
block_id=page_id,
children=[block_properties],
)
logging.info(f"Embed block {source} added to the page successfully!")
except Exception as e:
logging.info(f"Failed to add embed block {source} to the page. Error: {e}")
def add_image_to_page(self, page_id, image_url, embed=False):
if embed:
self.add_embed_to_page(page_id, image_url)
else:
self.load_image_to_page(page_id, image_url)
def query_database(self):
logging.info("Querying Notion database...")
results = []
next_cursor = None
while True:
# Query the database with the current cursor
response = self.notion.databases.query(
database_id=self.database_id,
start_cursor=next_cursor,
page_size=1000
)
# Append the results to the list
results.extend(response.get("results"))
# Check if there are more pages to retrieve
if not response.get("has_more"):
break
# Get the next cursor for the next page
next_cursor = response.get("next_cursor")
# save results
df = pd.DataFrame(results)
df.to_json("resources/notion_database.json", orient="records")
logging.info("Database query successful.")
return results
def extract_data_and_export_to_csv(self, results):
logging.info("Extracting data from query results...")
data = []
columns = results[0]["properties"].keys()
for page in results:
row = {}
row['id'] = page["id"]
for column in columns:
property_type = page["properties"][column]["type"]
if property_type in page["properties"][column]:
row[column] = page["properties"][column][property_type]
else:
row[column] = ""
data.append(row)
df = pd.DataFrame(data)
df.to_csv("resources/notion_database.csv", index=False,
quoting=csv.QUOTE_NONNUMERIC)
logging.info("Data extracted and exported to CSV.")
return df
def preprocess_df(self, df):
logging.info("Preprocessing DataFrame...")
df['Date'] = df['Date'].apply(lambda x: x['start'])
df['Date'] = pd.to_datetime(df['Date'], format='%Y-%m-%d')
df['Тема'] = df['Тема'].apply(lambda x: x[0]['name'])
df = df[['Date', 'id', 'Lesson','Grade','Тема']]
df = df.sort_values(by=['Date'])
logging.info("DataFrame preprocessed.")
print(df.head())
return df
def filter_df(self, df, inc = False):
logging.info("Filtering DataFrame...")
df_false = df[(df['Date'].dt.date == pd.to_datetime('today').date()) | (df['Lesson'] == False)]
filtered_df = df_false[df_false['Date'] <= pd.to_datetime('today')]
if not inc:
filtered_df = filtered_df[df_false['Date'] > df[df['Lesson']].iloc[-1]['Date']]
logging.info("DataFrame filtered.")
return filtered_df
def tick_lesson_checkbox(self, page_id):
logging.info(f"Ticking 'Lesson' checkbox for page {page_id}...")
update_properties = {
"Lesson": {
"checkbox": True
}
}
response = self.notion.pages.update(page_id, properties=update_properties)
if response:
logging.info(f"Successfully ticked 'Lesson' checkbox for page {page_id}!")
else:
logging.info(f"Failed to tick 'Lesson' checkbox for page {page_id}.")