dale_yxc/ragflow_python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

use minio to store uploaded files; build dialog server; (#16)

Browse Source 
* format code

* use minio to store uploaded files; build dialog server;
This commit is contained in:
KevinHuSh
2023-12-25 19:05:59 +08:00
committed by GitHub
parent d4fd138954
commit 3245107dc7
13 changed files with 520 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
src/api/doc_info.rs
View File

@@ -1,9 +1,9 @@
use std::collections::HashMap;
use std::io::Write;
use actix_multipart_extract::{File, Multipart, MultipartForm};
use actix_web::{get, HttpResponse, post, web};
use chrono::{Utc, FixedOffset};
use minio::s3::args::{BucketExistsArgs, MakeBucketArgs, UploadObjectArgs};
use std::io::BufReader;
use actix_multipart_extract::{ File, Multipart, MultipartForm };
use actix_web::{ HttpResponse, post, web };
use chrono::{ Utc, FixedOffset };
use minio::s3::args::{ BucketExistsArgs, MakeBucketArgs, PutObjectArgs };
use sea_orm::DbConn;
use crate::api::JsonResponse;
use crate::AppState;
@@ -12,9 +12,6 @@ use crate::errors::AppError;
use crate::service::doc_info::{ Mutation, Query };
use serde::Deserialize;
const BUCKET_NAME: &'static str = "docgpt-upload";
fn now() -> chrono::DateTime<FixedOffset> {
Utc::now().with_timezone(&FixedOffset::east_opt(3600 * 8).unwrap())
}
@@ -74,53 +71,71 @@ async fn upload(
) -> Result<HttpResponse, AppError> {
let uid = payload.uid;
let file_name = payload.file_field.name.as_str();
async fn add_number_to_filename(file_name: &str, conn:&DbConn, uid:i64, parent_id:i64) -> String {
async fn add_number_to_filename(
file_name: &str,
conn: &DbConn,
uid: i64,
parent_id: i64
) -> String {
let mut i = 0;
let mut new_file_name = file_name.to_string();
let arr: Vec<&str> = file_name.split(".").collect();
let suffix = String::from(arr[arr.len()-1]);
let preffix = arr[..arr.len()-1].join(".");
let mut docs = Query::find_doc_infos_by_name(conn, uid, &new_file_name, Some(parent_id)).await.unwrap();
while docs.len()>0 {
let suffix = String::from(arr[arr.len() - 1]);
let preffix = arr[..arr.len() - 1].join(".");
let mut docs = Query::find_doc_infos_by_name(
conn,
uid,
&new_file_name,
Some(parent_id)
).await.unwrap();
while docs.len() > 0 {
i += 1;
new_file_name = format!("{}_{}.{}", preffix, i, suffix);
docs = Query::find_doc_infos_by_name(conn, uid, &new_file_name, Some(parent_id)).await.unwrap();
docs = Query::find_doc_infos_by_name(
conn,
uid,
&new_file_name,
Some(parent_id)
).await.unwrap();
}
new_file_name
}
let fnm = add_number_to_filename(file_name, &data.conn, uid, payload.did).await;
let s3_client = &data.s3_client;
let bucket_name = format!("{}-upload", payload.uid);
let s3_client: &minio::s3::client::Client = &data.s3_client;
let buckets_exists = s3_client
.bucket_exists(&BucketExistsArgs::new(BUCKET_NAME)?)
.await?;
.bucket_exists(&BucketExistsArgs::new(&bucket_name).unwrap()).await
.unwrap();
if !buckets_exists {
s3_client
.make_bucket(&MakeBucketArgs::new(BUCKET_NAME)?)
.await?;
print!("Create bucket: {}", bucket_name.clone());
s3_client.make_bucket(&MakeBucketArgs::new(&bucket_name).unwrap()).await.unwrap();
} else {
print!("Existing bucket: {}", bucket_name.clone());
}
s3_client
.upload_object(
&mut UploadObjectArgs::new(
BUCKET_NAME,
fnm.as_str(),
format!("/{}/{}-{}", payload.uid, payload.did, fnm).as_str()
)?
)
.await?;
let location = format!("/{}/{}", payload.did, fnm);
print!("===>{}", location.clone());
s3_client.put_object(
&mut PutObjectArgs::new(
&bucket_name,
&location,
&mut BufReader::new(payload.file_field.bytes.as_slice()),
Some(payload.file_field.bytes.len()),
None
)?
).await?;
let location = format!("/{}/{}", BUCKET_NAME, fnm);
let doc = Mutation::create_doc_info(&data.conn, Model {
did:Default::default(),
uid: uid,
did: Default::default(),
uid: uid,
doc_name: fnm,
size: payload.file_field.bytes.len() as i64,
location,
r#type: "doc".to_string(),
created_at: now(),
updated_at: now(),
is_deleted:Default::default(),
is_deleted: Default::default(),
}).await?;
let _ = Mutation::place_doc(&data.conn, payload.did, doc.did.unwrap()).await?;

58
src/api/tag.rs
View File

@@ -1,58 +0,0 @@
use std::collections::HashMap;
use actix_web::{get, HttpResponse, post, web};
use actix_web::http::Error;
use crate::api::JsonResponse;
use crate::AppState;
use crate::entity::tag_info;
use crate::service::tag_info::{Mutation, Query};
#[post("/v1.0/create_tag")]
async fn create(model: web::Json<tag_info::Model>, data: web::Data<AppState>) -> Result<HttpResponse, Error> {
let model = Mutation::create_tag(&data.conn, model.into_inner()).await.unwrap();
let mut result = HashMap::new();
result.insert("tid", model.tid.unwrap());
let json_response = JsonResponse {
code: 200,
err: "".to_owned(),
data: result,
};
Ok(HttpResponse::Ok()
.content_type("application/json")
.body(serde_json::to_string(&json_response).unwrap()))
}
#[post("/v1.0/delete_tag")]
async fn delete(model: web::Json<tag_info::Model>, data: web::Data<AppState>) -> Result<HttpResponse, Error> {
let _ = Mutation::delete_tag(&data.conn, model.tid).await.unwrap();
let json_response = JsonResponse {
code: 200,
err: "".to_owned(),
data: (),
};
Ok(HttpResponse::Ok()
.content_type("application/json")
.body(serde_json::to_string(&json_response).unwrap()))
}
#[get("/v1.0/tags")]
async fn list(data: web::Data<AppState>) -> Result<HttpResponse, Error> {
let tags = Query::find_tag_infos(&data.conn).await.unwrap();
let mut result = HashMap::new();
result.insert("tags", tags);
let json_response = JsonResponse {
code: 200,
err: "".to_owned(),
data: result,
};
Ok(HttpResponse::Ok()
.content_type("application/json")
.body(serde_json::to_string(&json_response).unwrap()))
}

49
src/main.rs
View File

@@ -5,9 +5,9 @@ mod errors;
use std::env;
use actix_files::Files;
use actix_identity::{CookieIdentityPolicy, IdentityService, RequestIdentity};
use actix_identity::{ CookieIdentityPolicy, IdentityService, RequestIdentity };
use actix_session::CookieSession;
use actix_web::{web, App, HttpServer, middleware, Error};
use actix_web::{ web, App, HttpServer, middleware, Error };
use actix_web::cookie::time::Duration;
use actix_web::dev::ServiceRequest;
use actix_web::error::ErrorUnauthorized;
@@ -16,9 +16,9 @@ use listenfd::ListenFd;
use minio::s3::client::Client;
use minio::s3::creds::StaticProvider;
use minio::s3::http::BaseUrl;
use sea_orm::{Database, DatabaseConnection};
use migration::{Migrator, MigratorTrait};
use crate::errors::{AppError, UserError};
use sea_orm::{ Database, DatabaseConnection };
use migration::{ Migrator, MigratorTrait };
use crate::errors::{ AppError, UserError };
#[derive(Debug, Clone)]
struct AppState {
@@ -28,10 +28,10 @@ struct AppState {
pub(crate) async fn validator(
req: ServiceRequest,
credentials: BearerAuth,
credentials: BearerAuth
) -> Result<ServiceRequest, Error> {
if let Some(token) = req.get_identity() {
println!("{}, {}",credentials.token(), token);
println!("{}, {}", credentials.token(), token);
(credentials.token() == token)
.then(|| req)
.ok_or(ErrorUnauthorized(UserError::InvalidToken))
@@ -52,26 +52,25 @@ async fn main() -> Result<(), AppError> {
let port = env::var("PORT").expect("PORT is not set in .env file");
let server_url = format!("{host}:{port}");
let s3_base_url = env::var("S3_BASE_URL").expect("S3_BASE_URL is not set in .env file");
let s3_access_key = env::var("S3_ACCESS_KEY").expect("S3_ACCESS_KEY is not set in .env file");;
let s3_secret_key = env::var("S3_SECRET_KEY").expect("S3_SECRET_KEY is not set in .env file");;
let mut s3_base_url = env::var("MINIO_HOST").expect("MINIO_HOST is not set in .env file");
let s3_access_key = env::var("MINIO_USR").expect("MINIO_USR is not set in .env file");
let s3_secret_key = env::var("MINIO_PWD").expect("MINIO_PWD is not set in .env file");
if s3_base_url.find("http") != Some(0) {
s3_base_url = format!("http://{}", s3_base_url);
}
// establish connection to database and apply migrations
// -> create post table if not exists
let conn = Database::connect(&db_url).await.unwrap();
Migrator::up(&conn, None).await.unwrap();
let static_provider = StaticProvider::new(
s3_access_key.as_str(),
s3_secret_key.as_str(),
None,
);
let static_provider = StaticProvider::new(s3_access_key.as_str(), s3_secret_key.as_str(), None);
let s3_client = Client::new(
s3_base_url.parse::<BaseUrl>()?,
Some(Box::new(static_provider)),
None,
None,
Some(true)
)?;
let state = AppState { conn, s3_client };
@@ -82,18 +81,20 @@ async fn main() -> Result<(), AppError> {
App::new()
.service(Files::new("/static", "./static"))
.app_data(web::Data::new(state.clone()))
.wrap(IdentityService::new(
CookieIdentityPolicy::new(&[0; 32])
.name("auth-cookie")
.login_deadline(Duration::seconds(120))
.secure(false),
))
.wrap(
IdentityService::new(
CookieIdentityPolicy::new(&[0; 32])
.name("auth-cookie")
.login_deadline(Duration::seconds(120))
.secure(false)
)
)
.wrap(
CookieSession::signed(&[0; 32])
.name("session-cookie")
.secure(false)
// WARNING(alex): This uses the `time` crate, not `std::time`!
.expires_in_time(Duration::seconds(60)),
.expires_in_time(Duration::seconds(60))
)
.wrap(middleware::Logger::default())
.configure(init)
@@ -137,4 +138,4 @@ fn init(cfg: &mut web::ServiceConfig) {
cfg.service(api::user_info::login);
cfg.service(api::user_info::register);
cfg.service(api::user_info::setting);
}
}