Flask dummy with SSL
flask_manager.py
'''
$jsonData = @{
reqSystem = "model"
reqData = "llama3"
} | ConvertTo-Json
$utf8JsonData = [System.Text.Encoding]::UTF8.GetBytes($jsonData)
$responseUnicodeString = Invoke-WebRequest -Uri "http://11.11.11.11:8084/api/feinit" -Method POST -Body $utf8JsonData -ContentType "application/json"
$decodedString = [System.Text.RegularExpressions.Regex]::Unescape($responseUnicodeString)
$decodedString
3. 쿼리(파워쉘)
$jsonData = @{
reqUser = "넌 어떤 모델이야?"
reqResponse = ""
} | ConvertTo-Json
$utf8JsonData = [System.Text.Encoding]::UTF8.GetBytes($jsonData)
$responseUnicodeString = Invoke-WebRequest -Uri "http://11.11.11.11:80/api/femessage" -Method POST -Body $utf8JsonData -ContentType "application/json"
$decodedString = [System.Text.RegularExpressions.Regex]::Unescape($responseUnicodeString)
$decodedString
'''
'''
# JSON 데이터나 프로그래밍 언어에서 유니코드 문자를 안전하게 표현하기 위해 사용되는 유니코드 이스케이프(Unicode escape) 형식으로 인코딩된 문자열을 한글로 >디코딩
PS D:\Down> $decodedString = [System.Text.RegularExpressions.Regex]::Unescape($responseUnicodeString)
# 디코딩된 문자열 출력
PS D:\Down> $decodedString
'''
'''
pip install flask[async] flask_cors
'''
from flask import Flask, request, jsonify
from flask_cors import CORS
from typing import TypedDict
import asyncio
from util import CustomDict
def createCustomDict(_reqId: str, _reqSystem: str, _reqUser: str, _reqData: str, _reqResponse: str) -> CustomDict:
return CustomDict(
reqId=_reqId,
reqSystem=_reqSystem,
reqUser=_reqUser,
reqData=_reqData,
reqResponse=_reqResponse
)
class LangchainOllamaLLM:
def __init__(self, reqId: str, reqSystem: str, reqUser: str, reqData: str, reqResponse: str):
model_id = reqData
print(f"OllamaLLM.__init__: {model_id}")
async def answerLangchainOllamaLLM(self, custom_dict: CustomDict) -> CustomDict:
reqId = custom_dict['reqId']
reqSystem = custom_dict['reqSystem']
reqUser = custom_dict['reqUser']
reqData = custom_dict['reqData']
reqResponse = "이건 대답이다!!"
print(f"answerHuggingfaceLLM/reqResponse: {reqResponse}")
return createCustomDict(reqId, reqSystem, reqUser, reqData, reqResponse)
app = Flask(__name__)
CORS(app)
llmModel = LangchainOllamaLLM(
reqId=None,
reqSystem="model",
reqUser=None,
reqData="llama3",
reqResponse=None,
)
@app.route('/api/feinit', methods=['POST'])
async def llmInit():
data = request.get_json()
print(f"llmInit(): {data}")
llmModel.__init__(
reqId=data.get('reqId'),
reqSystem=data.get('reqSystem'),
reqUser=data.get('reqUser'),
reqData=data.get('reqData'),
reqResponse=data.get('reqResponse'),
)
return jsonify(data)
@app.route('/api/femessage', methods=['POST'])
async def receiveMessage():
data = request.get_json()
print(f"receiveMessage: {data}")
custom_dict = createCustomDict(
_reqId=data.get('reqId'),
_reqSystem=data.get('reqSystem'),
_reqUser=data.get('reqUser'),
_reqData=data.get('reqData'),
_reqResponse=data.get('reqResponse'),
)
print(f"receiveMessage()/custom_dict {custom_dict}")
print(f"receiveMessage()/custom_dict/reqUser {custom_dict['reqUser']}")
async def requestAnswer():
llmResponse = asyncio.create_task(llmModel.answerLangchainOllamaLLM(custom_dict))
answer = await llmResponse
return jsonify([answer])
return await requestAnswer()
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=80)
util.py
from typing import TypedDict
class CustomDict(TypedDict):
reqId: str # 사용안함
reqSystem: str # 사용안함
reqUser: str
reqData: str # 사용안함
reqResponse: str # 사용안함
안드로이드 클라이언트
network_security_config.xml 신규 생성 및 cert 등록
<?xml version="1.0" encoding="utf-8"?>
<network-security-config>
<domain-config>
<domain includeSubdomains="true">192.168.0.32</domain>
<trust-anchors>
<certificates src="@raw/cert" />
</trust-anchors>
</domain-config>
</network-security-config>
AndroidManifest.xml 에 network_security_config 추가
<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
xmlns:tools="http://schemas.android.com/tools" >
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<application
android:networkSecurityConfig="@xml/network_security_config"
android:allowBackup="true"
MainActivity.java에 인증 무력화
// MainActivity.java 파일
package com.example.restapitest;
import android.os.Bundle;
import com.google.android.material.snackbar.Snackbar;
import androidx.appcompat.app.AppCompatActivity;
import android.view.View;
import androidx.lifecycle.ViewModelProvider;
import androidx.navigation.NavController;
import androidx.navigation.Navigation;
import androidx.navigation.ui.AppBarConfiguration;
import androidx.navigation.ui.NavigationUI;
import com.example.restapitest.databinding.ActivityMainBinding;
import android.view.Menu;
import android.view.MenuItem;
import org.json.JSONObject;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
public class MainActivity extends AppCompatActivity {
private AppBarConfiguration appBarConfiguration;
private ActivityMainBinding binding;
private static final String SERVER_URL = "https://192.168.0.32:8084/api/femessage";
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
binding = ActivityMainBinding.inflate(getLayoutInflater());
setContentView(binding.getRoot());
setSupportActionBar(binding.toolbar);
NavController navController = Navigation.findNavController(this, R.id.nav_host_fragment_content_main);
appBarConfiguration = new AppBarConfiguration.Builder(navController.getGraph()).build();
NavigationUI.setupActionBarWithNavController(this, navController, appBarConfiguration);
binding.fab.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View view) {
Snackbar.make(view, "Replace with your own action", Snackbar.LENGTH_LONG)
.setAnchorView(R.id.fab)
.setAction("Action", null).show();
// JSON 데이터 생성 @@@
JSONObject jsonData = new JSONObject();
try {
jsonData.put("reqUser", "넌 어떤 모델이야?");
jsonData.put("reqResponse", "");
} catch (Exception e) {
e.printStackTrace();
}
// 서버 요청 및 응답 처리
sendPostRequest(jsonData.toString(), response -> {
// UI 업데이트는 메인 스레드에서 수행해야 함
SharedViewModel viewModel = new ViewModelProvider(MainActivity.this).get(SharedViewModel.class);
viewModel.setResponseText(response); // 응답 데이터 전달
});
}
});
}
private void sendPostRequest(String jsonData, ResponseCallback callback) {
// OkHttp 클라이언트 생성
// OkHttpClient client = new OkHttpClient();
OkHttpClient client = getUnsafeOkHttpClient();
// JSON 데이터 타입 설정
MediaType JSON = MediaType.get("application/json; charset=utf-8");
// 요청 바디 생성
RequestBody body = RequestBody.create(JSON, jsonData);
// 요청 객체 생성
Request request = new Request.Builder()
.url(SERVER_URL)
.post(body)
.build();
// 비동기 요청 실행
new Thread(() -> {
try {
Response response = client.newCall(request).execute();
if (response.isSuccessful() && response.body() != null) {
String responseString = response.body().string();
callback.onResponse(responseString); // 응답 데이터 전달
} else {
callback.onResponse("Error: " + response.code());
}
} catch (IOException e) {
e.printStackTrace();
callback.onResponse("Error: " + e.getMessage());
}
}).start();
}
// 모든 인증서와 호스트를 신뢰하는 OkHttpClient 생성
private OkHttpClient getUnsafeOkHttpClient() {
try {
// 모든 인증서를 신뢰하도록 TrustManager 설정
final TrustManager[] trustAllCerts = new TrustManager[]{
new X509TrustManager() {
@Override public void checkClientTrusted(java.security.cert.X509Certificate[] chain, String authType) {}
@Override public void checkServerTrusted(java.security.cert.X509Certificate[] chain, String authType) {}
@Override public java.security.cert.X509Certificate[] getAcceptedIssuers() { return new java.security.cert.X509Certificate[]{}; }
}
};
// SSLContext 초기화
final SSLContext sslContext = SSLContext.getInstance("SSL");
sslContext.init(null, trustAllCerts, new java.security.SecureRandom());
// SSLSocketFactory 생성
final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
// OkHttpClient 빌더에 적용
OkHttpClient.Builder builder = new OkHttpClient.Builder();
builder.sslSocketFactory(sslSocketFactory, (X509TrustManager) trustAllCerts[0]);
builder.hostnameVerifier((hostname, session) -> true); // 모든 호스트 허용
return builder
.connectTimeout(30, TimeUnit.SECONDS)
.readTimeout(30, TimeUnit.SECONDS)
.build();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
// 응답 콜백 인터페이스
interface ResponseCallback {
void onResponse(String response);
}
@Override
public boolean onCreateOptionsMenu(Menu menu) {
// Inflate the menu; this adds items to the action bar if it is present.
getMenuInflater().inflate(R.menu.menu_main, menu);
return true;
}
@Override
public boolean onOptionsItemSelected(MenuItem item) {
// Handle action bar item clicks here. The action bar will
// automatically handle clicks on the Home/Up button, so long
// as you specify a parent activity in AndroidManifest.xml.
int id = item.getItemId();
//noinspection SimplifiableIfStatement
if (id == R.id.action_settings) {
return true;
}
return super.onOptionsItemSelected(item);
}
@Override
public boolean onSupportNavigateUp() {
NavController navController = Navigation.findNavController(this, R.id.nav_host_fragment_content_main);
return NavigationUI.navigateUp(navController, appBarConfiguration)
|| super.onSupportNavigateUp();
}
}
FirstFragment.java 에서 서버의 응답을 화면에 텍스트로 출력
// FirstFragment.java file
package com.example.restapitest;
import android.os.Bundle;
import android.view.LayoutInflater;
import android.view.View;
import android.view.ViewGroup;
import androidx.annotation.NonNull;
import androidx.fragment.app.Fragment;
import androidx.lifecycle.ViewModelProvider;
import androidx.navigation.fragment.NavHostFragment;
import com.example.restapitest.databinding.FragmentFirstBinding;
public class FirstFragment extends Fragment {
private FragmentFirstBinding binding;
private SharedViewModel sharedViewModel;
@Override
public View onCreateView(
@NonNull LayoutInflater inflater, ViewGroup container,
Bundle savedInstanceState
) {
binding = FragmentFirstBinding.inflate(inflater, container, false);
return binding.getRoot();
}
public void onViewCreated(@NonNull View view, Bundle savedInstanceState) {
super.onViewCreated(view, savedInstanceState);
// ViewModel 가져오기
sharedViewModel = new ViewModelProvider(requireActivity()).get(SharedViewModel.class);
// LiveData 구독
sharedViewModel.getResponseText().observe(getViewLifecycleOwner(), text -> {
binding.textviewFirst.setText(text);
});
binding.buttonFirst.setOnClickListener(v ->
NavHostFragment.findNavController(FirstFragment.this)
.navigate(R.id.action_FirstFragment_to_SecondFragment)
);
}
@Override
public void onDestroyView() {
super.onDestroyView();
binding = null;
}
}
SharedViewModel.java 은 MainActivity에서 받은 응답 데이터를 FirstFragment 에 전달
// SharedViewModel.java
package com.example.restapitest;
import android.util.Log;
import androidx.lifecycle.LiveData;
import androidx.lifecycle.MutableLiveData;
import androidx.lifecycle.ViewModel;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
public class SharedViewModel extends ViewModel {
private final MutableLiveData<String> responseText = new MutableLiveData<>();
public void setResponseText(String text) {
try {
Gson gson = new GsonBuilder().setPrettyPrinting().create();
Object json = gson.fromJson(text, Object.class);
String prettyJson = gson.toJson(json);
Log.d("SharedViewModel", "setResponseText 파싱된 JSON:\n" + prettyJson);
text = prettyJson;
} catch (Exception e) {
Log.e("SharedViewModel", "JSON 파싱 실패", e);
}
responseText.postValue(text);
}
public LiveData<String> getResponseText() {
return responseText;
}
}
0 댓글