candifloss/flask_crud_app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upload and preview CSV

Browse Source
This commit is contained in:
Candifloss 2025-01-30 15:28:34 +05:30
parent ae6673f938
commit f7009d6e82
5 changed files with 37 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
app.py
View File

@ -1,8 +1,6 @@
from flask import Flask, request, render_template, redirect, abort, send_file
from flask import Flask, redirect
from sqlalchemy import exc
from models import *
import csv
from process_csv import process_csv # Import the CSV processing function
# Import the Blueprints
from routes.viewall import viewall_bp

29
process_csv.py
View File

@ -1,34 +1,29 @@
import csv
from io import TextIOWrapper
from models import db, Asset # Import your database and model
def process_csv(file):
def get_csv_data(file):
"""
Processes the uploaded CSV file and returns a list of Asset objects.
Processes the uploaded CSV file and returns a list of asset dictionaries.
Raises an exception if the file is invalid or data is incorrect.
"""
# Open and process the CSV file
csv_file = TextIOWrapper(file, encoding='utf-8')
reader = csv.DictReader(csv_file)
reader = csv.DictReader(csv_file, delimiter='|')
# Validate CSV headers
required_headers = ['assettag', 'hostname', 'warrantyfrom', 'status', 'staffnum']
if not all(header in reader.fieldnames for header in required_headers):
raise ValueError("CSV file must include headers: assettag, hostname, warrantyfrom, status, staffnum.")
# Process each row
# Extract data from the CSV file
assets = []
for row in reader:
# Validate required fields
if not row['assettag']:
raise ValueError("Missing 'assettag' in one or more rows.")
# Create Asset object
assets.append(Asset(
assettag=row['assettag'],
hostname=row['hostname'],
warrantyfrom=row['warrantyfrom'],
status=row['status'],
staffnum=row['staffnum']
))
assets.append({
'assettag': row['assettag'],
'hostname': row['hostname'],
'warrantyfrom': row['warrantyfrom'],
'status': row['status'],
'staffnum': row['staffnum']
})
return assets

25
routes/csvpreview.py
View File

@ -1,25 +0,0 @@
from flask import Blueprint, request, render_template
from models import Asset, db
viewcsv_bp = Blueprint('viewcsv', __name__)
@viewcsv_bp.route('/csv_preview', methods=['GET', 'POST'])
def csv_preview():
# Retrieve the CSV data from the session
csv_data = session.get('csv_data', [])
if request.method == 'POST':
# Save the data to the database
assets = [AssetTest(**row) for row in csv_data]
db.session.add_all(assets)
db.session.commit()
# Clear the session data
session.pop('csv_data', None)
# Redirect to view list page with success message
flash("Data saved to the database successfully!", 'success')
return redirect(url_for('view_list'))
# Render the preview page for GET requests
return render_template('csv_preview.html', csv_data=csv_data)

27
routes/upload.py
View File

@ -1,6 +1,10 @@
from flask import Blueprint, request, render_template, redirect
from models import Asset, db
import csv
from io import TextIOWrapper
from models import Asset
from process_csv import get_csv_data # Import the CSV processing function
# Create a Blueprint for upload
upload_bp = Blueprint('uploadcsv', __name__)
@upload_bp.route('/uploadcsv', methods=['GET', 'POST'])
@ -8,30 +12,25 @@ def upload_file():
if request.method == 'POST':
# Check if a file was uploaded
if 'file' not in request.files:
flash("No file uploaded.", 'error')
return redirect(url_for('upload_file'))
return redirect('uploadcsv')
file = request.files['file']
# Check if the file is a CSV
if file.filename == '' or not file.filename.endswith('.csv'):
flash("Please upload a valid CSV file.", 'error')
return redirect(url_for('upload_file'))
return redirect('uploadcsv')
try:
# Process the CSV file
assets = process_csv(file)
# Extract data from the CSV file
assets = get_csv_data(file)
# Store the processed data in the session for preview
session['csv_data'] = [asset.__dict__ for asset in assets]
# Redirect to preview page
return redirect(url_for('csv_preview'))
# Redirect to preview page with the CSV data
return render_template('csv_preview.html', assets=assets)
except Exception as e:
# Handle errors during file processing
flash(f"Error processing CSV file: {str(e)}", 'error')
return redirect(url_for('upload_file'))
print(f"Error processing CSV file: {str(e)}")
return redirect('uploadcsv')
# Render the upload page for GET requests
return render_template('upload.html')

29
templates/csv_preview.html
View File

@ -3,19 +3,10 @@
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Preview CSV Data</title>
<title>CSV Preview</title>
</head>
<body>
<h1>Preview CSV Data</h1>
<!-- Display flash messages -->
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<p style="color: {% if category == 'error' %}red{% else %}green{% endif %};">{{ message }}</p>
{% endfor %}
{% endif %}
{% endwith %}
<h1>CSV Preview</h1>
<!-- Display CSV data in a table -->
<table border="1">
@ -29,20 +20,20 @@
</tr>
</thead>
<tbody>
{% for row in csv_data %}
{% for asset in assets %}
<tr>
<td>{{ row.assettag }}</td>
<td>{{ row.hostname }}</td>
<td>{{ row.warrantyfrom }}</td>
<td>{{ row.status }}</td>
<td>{{ row.staffnum }}</td>
<td>{{ asset.assettag }}</td>
<td>{{ asset.hostname }}</td>
<td>{{ asset.warrantyfrom }}</td>
<td>{{ asset.status }}</td>
<td>{{ asset.staffnum }}</td>
</tr>
{% endfor %}
</tbody>
</table>
<!-- Confirmation form -->
<form method="POST">
<!-- Form to confirm and save data -->
<form action="/confirm_save" method="POST">
<button type="submit">Confirm and Save to Database</button>
</form>
</body>