import requests
from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce import exception
from baidubce.services import bos
from baidubce.services.bos import canned_acl
from baidubce.services.bos.bos_client import BosClient

# bos
bos_host = "bj.bcebos.com"
access_key_id = "${access_key}"
secret_access_key = "${secret_key}"
config = BceClientConfiguration(credentials=BceCredentials(access_key_id, secret_access_key), endpoint = bos_host)
bos_client = BosClient(config)
bucket_name = "${bucket_name}"
object_prefix = "${object_prefix}"     # example: /upload/

def process_row(row):
    url = row.url
    name = row.name
    local_filename = name + ".pdf"

    response = requests.get(url, stream=True)
    response.raise_for_status()
    with open(local_filename, 'wb') as file_handle:
        for chunk in response.iter_content(chunk_size=8192):
            file_handle.write(chunk)
    
    object_key = object_prefix + local_filename
    data = open(local_filename, 'rb')
    bos_client.put_object_from_file(bucket_name, object_key, local_filename)
    bos_location = f"bos://{bucket_name}{object_key}"
    return (row.id, row.name, row.url, bos_location)


df = spark.sql("select * from sanling.pdf_files")
processed_rdd = df.rdd.map(process_row)
processed_df = spark.createDataFrame(processed_rdd, df.schema)

processed_df.write.mode("overwrite").insertInto("sanling.pdf_files")

spark.sql("select * from sanling.pdf_files").show()

spark.hadoop.edap.endpoint=http://edap.bj.baidubce.com
spark.hadoop.fs.AbstractFileSystem.bos.impl=org.apache.hadoop.fs.bos.BaiduBosFileSystemAdapter
spark.hadoop.hive.metastore.client.class=com.baidubce.edap.catalog.metastore.EdapCatalogHiveClient
spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog
spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension