euronext-scraper/scraper.py

111 lines
2.6 KiB
Python
Raw Normal View History

2023-12-16 15:45:40 +00:00
import argparse
import json
import sys
from dataclasses import dataclass
from pathlib import Path
import sqlite3
from typing import List, NamedTuple, Dict
import requests
DEFAULT_CONFIG_PATH = Path("config.json")
EURONEXT_BASE_URL = "https://live.euronext.com/intraday_chart/getChartData"
@dataclass
class ProgramArgs:
config_path: Path
class Action(NamedTuple):
name: str
code: str
class StoredAction(NamedTuple):
name: str
value: float
date: str
@dataclass
class Config:
actions: List[Action]
db_path: Path
def load_config(config_path: Path) -> Config:
with open(config_path) as conf_file:
raw_conf = json.load(conf_file)
conf = Config(
actions=[Action(name=a["name"], code=a["code"]) for a in raw_conf["actions"]],
db_path=Path(raw_conf["db"]["path"]),
)
return conf
def get_last_value_for_action(action: Action) -> StoredAction:
url = f"{EURONEXT_BASE_URL}/{action.code}/intraday"
resp = requests.get(url)
resp_json = resp.json()
if len(resp_json) < 1:
raise ValueError("Empty list of values")
last_value = resp_json[-1]
try:
return StoredAction(
name=action.name,
value=last_value["price"],
date=last_value["time"],
)
except KeyError as exc:
raise ValueError("Invalid format for response") from exc
def save_values_to_db(db_path: Path, values: List[StoredAction]) -> None:
con = sqlite3.connect(db_path)
cur = con.cursor()
cur.execute("DELETE FROM actions")
query_data = [(k.name, k.value, k.date) for k in values]
cur.executemany(
"INSERT INTO actions (name, value, date) VALUES(?, ?, ?)", query_data
)
con.commit()
def main():
args = parse_args()
conf = load_config(args.config_path)
res = []
for action in conf.actions:
try:
value = get_last_value_for_action(action)
res.append(value)
except ValueError as exc:
print(f"Got error {exc} for action {action.name}")
save_values_to_db(conf.db_path, res)
def parse_args() -> ProgramArgs:
parser = argparse.ArgumentParser(
prog="Euronext scraper", description="Scraper for euronext action values"
)
parser.add_argument(
"--config",
"-c",
type=Path,
help="Path for the configuration file",
default=DEFAULT_CONFIG_PATH,
)
args = parser.parse_args()
return ProgramArgs(
config_path=args.config,
)
if __name__ == "__main__":
try:
main()
except Exception as e:
print(f"An error occured: {e}")
sys.exit(1)