中级

staging 流水行数

未完成

中级参考代码结构已给出，请填写 ____ 处


import json
import os
import sqlite3
from pathlib import Path

import pandas as pd

def _root():
    env = os.environ.get(____)
    if env:
        return Path(____)
    return Path(____)

def _load_csv(____):
    p = _root() / name
    if p.is_file():
        return pd.read_csv(____).fillna(____)
    return pd.read_csv(____).fillna(____)

def _load_json(____):
    p = _root() / name
    if p.is_file():
        return json.loads(p.read_text(____))
    return pd.read_json(____).to_dict(____)

def _db():
    con = sqlite3.connect(____)
    con.executescript(____)
    return con

def solve():
    con = _db()
    raw = _load_json(____)
    txn = pd.DataFrame([
        {
            '____': r['____'], '____': r['____'], '____': r['____'], '____': r['____'],
            '____': r.get(____).get(____),
            '____': r.get(____).get(____),
        }
        for r in raw
    ])
    txn.to_sql(____)
    return con.execute(____).fetchone()[____]

示例

输入

solve()

期望输出

Python 代码 🔒 登录后使用

import json
import os
import sqlite3
from pathlib import Path

import pandas as pd

def _root():
    env = os.environ.get("ETL_FIXTURE_DIR")
    if env:
        return Path(env)
    return Path("/static/data/etl/bank-east-1104")

def _load_csv(name):
    p = _root() / name
    if p.is_file():
        return pd.read_csv(p, dtype=str).fillna("")
    return pd.read_csv(f"/static/data/etl/bank-east-1104/{name}", dtype=str).fillna("")

def _load_json(name):
    p = _root() / name
    if p.is_file():
        return json.loads(p.read_text(encoding="utf-8"))
    return pd.read_json(f"/static/data/etl/bank-east-1104/{name}").to_dict(orient="records")

def _db():
    con = sqlite3.connect(":memory:")
    con.executescript("""
    CREATE TABLE stg_org(nbjgh TEXT, yhjgmc TEXT, jrxkzh TEXT);
    CREATE TABLE stg_cif_indiv(cust_no TEXT, name TEXT, id_type TEXT, id_no TEXT, update_dt TEXT);
    CREATE TABLE stg_cif_corp(cust_no TEXT, name TEXT, uscc TEXT, update_dt TEXT);
    CREATE TABLE stg_deposit_acct(zh TEXT, cust_no TEXT, cust_type TEXT, nbjgh TEXT, zhye REAL, bz TEXT, biz_date TEXT, status TEXT);
    CREATE TABLE stg_deposit_txn(zh TEXT, jyrq TEXT, jyje REAL, jyfx TEXT, jydsxm TEXT, jydszh TEXT);
    CREATE TABLE stg_loan_contract(htbh TEXT, cust_no TEXT, cust_type TEXT, htje REAL, qsrq TEXT, dqrq TEXT, biz_date TEXT);
    CREATE TABLE stg_loan_acct(dkfhzh TEXT, htbh TEXT, cust_no TEXT, cust_type TEXT, dkye REAL, wjfl TEXT, biz_date TEXT);
    CREATE TABLE stg_gl_balance(km TEXT, kmmc TEXT, ye REAL, bgq TEXT);
    CREATE TABLE JGXXB(nbjgh TEXT, yhjgmc TEXT, jrxkzh TEXT, bgq TEXT);
    CREATE TABLE GRJCXXB(khtybh TEXT, khxm TEXT, zjlb TEXT, zjhm TEXT, bgq TEXT);
    CREATE TABLE DGKHXXB(khtybh TEXT, khmc TEXT, tyshxydm TEXT, bgq TEXT);
    CREATE TABLE GRCKFHZ(zh TEXT, khtybh TEXT, nbjgh TEXT, zhye REAL, bz TEXT, bgq TEXT);
    CREATE TABLE DGCKFHZ(zh TEXT, khtybh TEXT, nbjgh TEXT, zhye REAL, bz TEXT, bgq TEXT);
    CREATE TABLE GRCKFHZMX(zh TEXT, jyrq TEXT, jyje REAL, jyfx TEXT, jydsxm TEXT, jydszh TEXT);
    CREATE TABLE DGCKFHZMX(zh TEXT, jyrq TEXT, jyje REAL, jyfx TEXT, jydsxm TEXT, jydszh TEXT);
    CREATE TABLE XDHTB(htbh TEXT, khtybh TEXT, htje REAL, qsrq TEXT, dqrq TEXT, bgq TEXT);
    CREATE TABLE GRXDFHZ(dkfhzh TEXT, htbh TEXT, khtybh TEXT, dkye REAL, wjfl TEXT, bgq TEXT);
    CREATE TABLE DGXDFHZ(dkfhzh TEXT, htbh TEXT, khtybh TEXT, dkye REAL, wjfl TEXT, bgq TEXT);
    CREATE TABLE rpt_g01(xmdm TEXT, xmmc TEXT, ye REAL, bgq TEXT);
    CREATE TABLE rpt_loan_5class(wjfl TEXT, ye REAL, bgq TEXT);
    CREATE TABLE rpt_tieout(rule TEXT, east REAL, g01 REAL, gl REAL, diff REAL, pass INTEGER, bgq TEXT);
    CREATE TABLE dq_reject(rule TEXT, tbl TEXT, pk TEXT, reason TEXT, bgq TEXT);
    """)
    return con

def solve():
    return 0

🔒

登录后即可练习

注册免费账号，在浏览器中直接运行 Python 代码

登录免费注册

staging 流水行数

示例

登录后即可练习

升级 VIP

联系客服