Main Scripts
Refund & Exchange & Split Order Script

20min
Introduction
This script is designed to send emails to clients when they request a refund or exchange. it also send emasils to clients whose orders has been split, and will arrrive at different times. It first makes a SQL query to Linnworks to fetch data about the requested refund or exchange. Once it has the necessary data, the script sends an email to the client using a professional email template. The script is written in Python and uses the SMTPlib library to send emails.
Installation
Clone the repository and install the dependencies
Shell
git clone https://github.com/E-MPS-CO-UK/Refund-Mail
cd refund-mail
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
git clone https://github.com/E-MPS-CO-UK/Refund-Mail
cd refund-mail
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
﻿
Create a .env file and add the get the keys from the Linnworks developer account﻿
Shell
ApplicatonId=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
ApplicationSecret=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
Token=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
ApplicatonId=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
ApplicationSecret=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
Token=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
﻿
Running the script
Shell
python3 main.py
python3 main.py
﻿
Code Structure
Script Structure
Shell
.
├── LICENSE
├── README.md
├── Refund
│   ├── __init__.py
│   ├── config.py
│   ├── logger.py
│   ├── refund_script.py
│   ├── sql_scripts
│   │   ├── __init__.py
│   │   ├── exchange_script.py
│   │   ├── refund_script.py
│   │   └── split_script.py
│   ├── templates
│   │   ├── email_template_Stock.html
│   │   ├── email_template_exchange.html
│   │   ├── email_template_exchange_international.html
│   │   ├── email_template_oos.html
│   │   ├── email_template_refund.html
│   │   └── email_template_split_order.html
│   └── utils.py
├── config.ini
├── main.py
├── pyproject.toml
├── requirements.txt
├── tests
│   ├── __init__.py
│   ├── test_refund_script.py
│   └── test_utils.py
├── towerlondonRefunds.db
└── tox.ini
.
├── LICENSE
├── README.md
├── Refund
│   ├── __init__.py
│   ├── config.py
│   ├── logger.py
│   ├── refund_script.py
│   ├── sql_scripts
│   │   ├── __init__.py
│   │   ├── exchange_script.py
│   │   ├── refund_script.py
│   │   └── split_script.py
│   ├── templates
│   │   ├── email_template_Stock.html
│   │   ├── email_template_exchange.html
│   │   ├── email_template_exchange_international.html
│   │   ├── email_template_oos.html
│   │   ├── email_template_refund.html
│   │   └── email_template_split_order.html
│   └── utils.py
├── config.ini
├── main.py
├── pyproject.toml
├── requirements.txt
├── tests
│   ├── __init__.py
│   ├── test_refund_script.py
│   └── test_utils.py
├── towerlondonRefunds.db
└── tox.ini
﻿
Code Overview
Database
The database is created automatically when you run the code, if the database doesn’t already exist. It is used to store order data and retrieve processed order data on demand. The database used in this script is SQLite.
Python
conn = sqlite3.connect('towerlondonRefunds.db')
cursor = conn.cursor()
cursor.execute(RETURN_TABLE_QUERY)
conn.commit()
conn = sqlite3.connect('towerlondonRefunds.db')
cursor = conn.cursor()
cursor.execute(RETURN_TABLE_QUERY)
conn.commit()
﻿
The RETURN_TABLE_QUERY also creates the table if does not already exist.
Text
CREATE TABLE IF NOT EXISTS Refunds(
ID INTEGER PRIMARY KEY AUTOINCREMENT,
CustomerName TEXT)
CREATE TABLE IF NOT EXISTS Refunds(
ID INTEGER PRIMARY KEY AUTOINCREMENT,
CustomerName TEXT)
﻿
Refund Script Module
The Linnapi class is designed to interact with the Linnworks API for fetching and processing split, exchange and refund orders data.
Token Management: This method is responsible for obtaining a new authentication token to interact with the Linnworks API. It ensures that a valid token is always available.
Python
    def get_new_token(self) -> None:
        response = requests.post(self.token_url,params=self.refresh_data,
                                 timeout=30)
        if response.status_code != 200:
            raise HTTPError("Failed to get new token. Status code:"
                            f"{response.status_code}")
        response_data = response.json()
        new_access_token = response_data['Token']
        self.headers["Authorization"] = new_access_token
    def get_new_token(self) -> None:
        response = requests.post(self.token_url,params=self.refresh_data,
                                 timeout=30)
        if response.status_code != 200:
            raise HTTPError("Failed to get new token. Status code:"
                            f"{response.status_code}")
        response_data = response.json()
        new_access_token = response_data['Token']
        self.headers["Authorization"] = new_access_token
﻿
The method is used to fetch the access token and add it to the headers. If the token has expired and has not been refreshed, it is fetched again using the get_data_from_api method
Data Retrieval: This method fetches data from the Linnworks API using a POST request and returns the JSON response. It also handles token refresh if the current token is invalid. it mostly acts as a helper function to fetch data
Python
def get_data_from_api(self, api_url: str, script: str, 
                      headers: dict[str, str]) -> Dict[str, Any]:
    r = requests.post(api_url, data={"script": script}, headers=headers,
                      timeout=60)
    if r.status_code != 200:
        self.get_new_token()
        r = requests.post(api_url, data={"script": script}, headers=headers, 
                          timeout=60)
    response: Dict[str, Any] = r.json()
    return response
def get_data_from_api(self, api_url: str, script: str, 
                      headers: dict[str, str]) -> Dict[str, Any]:
    r = requests.post(api_url, data={"script": script}, headers=headers,
                      timeout=60)
    if r.status_code != 200:
        self.get_new_token()
        r = requests.post(api_url, data={"script": script}, headers=headers, 
                          timeout=60)
    response: Dict[str, Any] = r.json()
    return response
﻿
insert_data_into_db: This method inserts data into an SQLite database using the provided cursor. It is used to store refund and exchange data. For more info on the sqlite python documentation visit this LINK﻿
process_data - The process_data function is a part of a data pipeline that extracts, transforms, and loads (ETL) data into a SQLite database.
Extracts Data: It uses a custom script query (data_query) to retrieve data from an external source via an API. The API URL is constructed using the base URL (self.apiurl) and the endpoint (‘/Dashboards/ExecuteCustomScriptQuery’).
Transforms Data: The retrieved data is transformed into a pandas DataFrame for easier manipulation. It then iterates over each row of the DataFrame, checking if the record already exists in the database. If not, it inserts the new record into the database.
Loads Data: The transformed data is loaded into a SQLite database (towerlondonRefunds.db). The function creates a table in the database using the provided SQL query (table_query), and inserts data into the specified table (table_name).
Email Notifications: After loading the data into the database, it retrieves all unprocessed records (where processed = 0) from the table and triggers email notifications for each record using the provided email function (email_func).
the code snippet shows how the function works.
Python
data = self.get_data_from_api(api_url, script, self.headers)
df = pd.DataFrame(data['Results'])
for index, row in df.iterrows():
    cursor.execute(
        f"SELECT * FROM {table_name} WHERE OrderID=?", (row['OrderID'],))
    if cursor.fetchone() is None:
        values = [row[col] for col in columns]
        self.insert_data_into_db(cursor, table_name, columns, values)
main_log.info('%s orders inserted for table %s', len(df), table_name)
conn.commit()
existing_data = pd.read_sql(
    f"SELECT * FROM {table_name} WHERE processed = 0", conn)
main_log.info('orders to be processed for table name %s are %s',
              table_name, len(existing_data))
for index, row in existing_data.iterrows():
    email_func(row.to_dict(), cursor, conn)
conn.close()
data = self.get_data_from_api(api_url, script, self.headers)
df = pd.DataFrame(data['Results'])
for index, row in df.iterrows():
    cursor.execute(
        f"SELECT * FROM {table_name} WHERE OrderID=?", (row['OrderID'],))
    if cursor.fetchone() is None:
        values = [row[col] for col in columns]
        self.insert_data_into_db(cursor, table_name, columns, values)
main_log.info('%s orders inserted for table %s', len(df), table_name)
conn.commit()
existing_data = pd.read_sql(
    f"SELECT * FROM {table_name} WHERE processed = 0", conn)
main_log.info('orders to be processed for table name %s are %s',
              table_name, len(existing_data))
for index, row in existing_data.iterrows():
    email_func(row.to_dict(), cursor, conn)
conn.close()
﻿
refund_data, exchange_data, and split_order - Each of these methods processes a specific type of order (refunds, exchanges, and split orders respectively) by calling the process_data function with parameters specific to each order type. These parameters include the SQL queries for creating the table and retrieving data, the table name, column names, and a function for sending emails specific to the order type (send_refund_mail, send_exchange_mail, or send_split_mail).
Utils Module
This Python module provides a set of helper methods for sending emails. It is designed to streamline the process of sending various types of emails, including refund notifications and order exchange notifications.
send_email_template - This function sends an email using a provided template file and dynamic data. It handles the process of replacing placeholders in the template with the dynamic values and sending the email.
Parameters are:
template_file (str): Path to the template file.
name (str): Recipient's name.
email (str): Recipient's email address.
subject (str): Email subject.
bcc (str): Bcc email address.
data (dict): Dictionary containing dynamic data for the template.
caller: function calling the send_email_template function.
The code snippet shoes how data is filled to the template File, and then an email is sent to the client.
Python
# Read in the template file
with open(template_file, "r", encoding='utf-8') as f:
            template = f.read()
# Replace placeholders with dynamic values
for key, value in data.items():
    template = template.replace(f"{{{key}}}", str(value))
# Create the email message
msg = MIMEMultipart()
msg["From"] = config.client_email
msg["To"] = email
msg["Subject"] = subject
msg["Bcc"] = bcc
msg.attach(MIMEText(template, 'html'))
# Send the email
server = smtplib.SMTP(config.client_host, int(config.client_port))
server.starttls()
server.login(config.client_email, config.client_password)
server.send_message(msg)
main_log.info("sending %s email to %s", subject, 'iandan')
server.quit()
# Read in the template file
with open(template_file, "r", encoding='utf-8') as f:
            template = f.read()
# Replace placeholders with dynamic values
for key, value in data.items():
    template = template.replace(f"{{{key}}}", str(value))
# Create the email message
msg = MIMEMultipart()
msg["From"] = config.client_email
msg["To"] = email
msg["Subject"] = subject
msg["Bcc"] = bcc
msg.attach(MIMEText(template, 'html'))
# Send the email
server = smtplib.SMTP(config.client_host, int(config.client_port))
server.starttls()
server.login(config.client_email, config.client_password)
server.send_message(msg)
main_log.info("sending %s email to %s", subject, 'iandan')
server.quit()
﻿
create_email_data - This function is designed to prepare the data needed for sending an email. It takes a dictionary of row data (row_dict) and a template file path (template_file) as inputs.
update_order_status - This function is used update the OrderId status in the Database after an e-mail has been sent. processed is changed from 0 to 1.
prepare_email - this function is a utility function used to prepare and send an email using data from a SQLite database and a specified template file. Here’s a brief description of its functionality and parameters:
Functionality: The function first calls the create_email_data function to prepare the email data using the provided row data and template file. It then sends the email using the send_email function, and updates the order status in the database using the update_order_status function. If any error occurs during this process, it logs the error using main_log.error.
code snippet:
Python
template_file, data = create_email_data(row_dict, template_file)
send_email(template_file,
           data['Full Name'],
           data['Email Address'],
           subject,
           f'{config.bcc_email1},{config.bcc_email2},{config.bcc_email3}',
           data, function_name)
update_order_status(cursor, conn, table_name, row_dict['OrderID'])
template_file, data = create_email_data(row_dict, template_file)
send_email(template_file,
           data['Full Name'],
           data['Email Address'],
           subject,
           f'{config.bcc_email1},{config.bcc_email2},{config.bcc_email3}',
           data, function_name)
update_order_status(cursor, conn, table_name, row_dict['OrderID'])
﻿
send_refund_email - The function is used to send an email notification based on a refund or out-of-stock reason.
send_exchange_mail - The function is used to send an email notification for order exchange.
send_split_mail - The function is used to send an email notification for an order that has been Split
They are different but serve the same purpose.
Example of the send_exchange_mail
Python
if row_dict['country'] == 'United Kingdom':
        template_file = config.template_exchange
else:
    template_file = config.template_exchange_international
prepare_email(row_dict, cursor, conn, template_file,
              "Order Exchange Notification", "exchanges",
              "send_exchange_mail")
if row_dict['country'] == 'United Kingdom':
        template_file = config.template_exchange
else:
    template_file = config.template_exchange_international
prepare_email(row_dict, cursor, conn, template_file,
              "Order Exchange Notification", "exchanges",
              "send_exchange_mail")
﻿
Config Module
The Configuration Module serves as a centralized location for various essential parameters and credentials. It includes:
File Paths: Paths to files that the application needs to access.
API Keys: Keys required to authenticate with various APIs.
Email Credentials: Credentials needed to send emails from the application.
Logfile Paths: Locations of log files for application debugging and record-keeping.
Function Configurations: Settings and parameters for various functions within the application.
Templates
The template files are stored in the templates folder. The template files are in html format and contain placeholders for dynamic values. The placeholders are replaced with dynamic values before the email is sent. The template files are passed to the send_email_template function as parameters. The function then replaces the placeholders with the dynamic values and sends the email to the client.
Tests
The Test suite in this script has been written with Pytest
To run the Tests run
Shell
pytest tests/
pytest tests/
﻿
The Schedule Process
This script run as a PM2 process on the server. the process is scheduled to run as specified below. I use the Schedule Library to schedule the process.
Python
def main() -> None:
    linn = Linnapi()
    parser = argparse.ArgumentParser(description='Run scheduled tasks.')
    parser.add_argument('--getrefund-data', type=int, default=40,
                        help='Interval (in minutes) for running the'
                             'refund function')
    parser.add_argument('--getexchange-data', type=int, default=50,
                        help='Interval (in minutes) for running the'
                             'exchange_data function')
    args = parser.parse_args()
    schedule.every(args.getrefund_data).minutes.do(linn.return_data)
    schedule.every(args.getexchange_data).minutes.do(linn.exchange_data)

    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    main()
def main() -> None:
    linn = Linnapi()
    parser = argparse.ArgumentParser(description='Run scheduled tasks.')
    parser.add_argument('--getrefund-data', type=int, default=40,
                        help='Interval (in minutes) for running the'
                             'refund function')
    parser.add_argument('--getexchange-data', type=int, default=50,
                        help='Interval (in minutes) for running the'
                             'exchange_data function')
    args = parser.parse_args()
    schedule.every(args.getrefund_data).minutes.do(linn.return_data)
    schedule.every(args.getexchange_data).minutes.do(linn.exchange_data)

    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    main()
﻿
This is a general overview of the Refund, Exchange & Split Order Mail. With the Information provided you will be able to run the script seamlessly.
﻿
Updated 25 Mar 2024
Did this page help you?
Scripts
Dropshipping Script